OLAP增加Datapath Telemetry Record

Clickhouse最新全量建表语句/Clickhouse_TSG_建表语句.sql

@@ -4251,3 +4251,67 @@ SELECT
 FROM tsg_galaxy_v3.session_record_local
 WHERE empty(monitor_rule_list) = 0
 ;
+CREATE TABLE IF NOT EXISTS tsg_galaxy_v3.datapath_telemetry_record_local on cluster ck_cluster (
+    log_id UInt64,
+    recv_time Int64,
+    vsys_id Int32,
+    timestamp_us UInt64,
+    job_id String,
+    sled_ip String,
+    device_group String,
+    traffic_link_id Int32,
+    source_ip String,
+    source_port Int32,
+    destination_ip String,
+    destination_port Int32,
+    packet String,
+    packet_length Int32,
+    measurements String
+)
+ENGINE = MergeTree
+PARTITION BY toYYYYMMDD(toDate(recv_time))
+ORDER BY (vsys_id,job_id,recv_time,timestamp_us);
+
+CREATE TABLE IF NOT EXISTS tsg_galaxy_v3.datapath_telemetry_record on cluster ck_cluster (
+    log_id UInt64,
+    recv_time Int64,
+    vsys_id Int32,
+    timestamp_us UInt64,
+    job_id String,
+    sled_ip String,
+    device_group String,
+    traffic_link_id Int32,
+    source_ip String,
+    source_port Int32,
+    destination_ip String,
+    destination_port Int32,
+    packet String,
+    packet_length Int32,
+    measurements String
+)
+ENGINE = Distributed('ck_cluster',
+ 'tsg_galaxy_v3',
+ 'datapath_telemetry_record_local',
+ rand());
+ 
+ CREATE TABLE IF NOT EXISTS tsg_galaxy_v3.datapath_telemetry_record on cluster ck_query (
+    log_id UInt64,
+    recv_time Int64,
+    vsys_id Int32,
+    timestamp_us UInt64,
+    job_id String,
+    sled_ip String,
+    device_group String,
+    traffic_link_id Int32,
+    source_ip String,
+    source_port Int32,
+    destination_ip String,
+    destination_port Int32,
+    packet String,
+    packet_length Int32,
+    measurements String
+)
+ENGINE = Distributed('ck_cluster',
+ 'tsg_galaxy_v3',
+ 'datapath_telemetry_record_local',
+ rand());

Groot-Stream 最新全量配置模版/TSG/集中式/etl_datapath_telemetry_record_kafka_to_clickhouse

@@ -0,0 +1,71 @@
+sources:
+  kafka_source:
+    type: kafka
+    properties:
+      topic: DATAPATH-TELEMETRY-RECORD
+      kafka.bootstrap.servers: 192.168.44.11:9094,192.168.44.13:9094,192.168.44.14:9094,192.168.44.15:9094,192.168.44.16:9094
+      kafka.session.timeout.ms: 60000
+      kafka.max.poll.records: 3000
+      kafka.max.partition.fetch.bytes: 31457280
+      #kafka.security.protocol: SSL
+      #kafka.ssl.endpoint.identification.algorithm: ""
+      #kafka.ssl.keystore.location: /data/tsg/olap/flink/topology/data/keystore.jks
+      #kafka.ssl.keystore.password: 86cf0e2ffba3f541a6c6761313e5cc7e
+      #kafka.ssl.truststore.location: /data/tsg/olap/flink/topology/data/truststore.jks
+      #kafka.ssl.truststore.password: 86cf0e2ffba3f541a6c6761313e5cc7e
+      #kafka.ssl.key.password: 86cf0e2ffba3f541a6c6761313e5cc7e
+	  kafka.security.protocol: SASL_PLAINTEXT
+      kafka.sasl.mechanism: PLAIN
+      kafka.sasl.jaas.config: 454f65ea6eef1256e3067104f82730e737b68959560966b811e7ff364116b03124917eb2b0f3596f14733aa29ebad9352644ce1a5c85991c6f01ba8a5e8f177a7ff0b2d3889a424249967b3870b50993d9644f239f0de82cdb13bdb502959e16afadffa49ef1e1d2b9c9b5113e619817
+      kafka.group.id: etl_datapath_telemetry_record_kafka_to_clickhouse-20230125
+      kafka.auto.offset.reset: latest
+      format: msgpack
+
+processing_pipelines:
+  etl_processor: # [object] Processing Pipeline
+    type: com.geedgenetworks.core.processor.projection.ProjectionProcessorImpl
+    remove_fields:
+    output_fields:
+    functions:
+      - function: SNOWFLAKE_ID
+        lookup_fields: [ '' ]
+        output_fields: [ log_id ]
+        parameters:
+          data_center_id_num: 1
+      - function: UNIX_TIMESTAMP_CONVERTER
+        lookup_fields: [ __timestamp ]
+        output_fields: [ recv_time ]
+        parameters:
+          precision: seconds
+      - function: BASE64_ENCODE_TO_STRING
+        output_fields: [ packet ]
+        parameters:
+          value_field: packet
+
+sinks:
+  clickhouse_sink:
+    type: clickhouse
+    properties:
+      host: 192.168.44.13:9001,192.168.44.14:9001,192.168.44.15:9001,192.168.44.16:9001
+      table: tsg_galaxy_v3.datapath_telemetry_record_local
+      batch.size: 5000
+      batch.interval: 30s
+      connection.user: e54c9568586180eede1506eecf3574e9
+      connection.password: 86cf0e2ffba3f541a6c6761313e5cc7e
+
+
+application:
+
+  env: # [object] Environment Variables
+    name: etl_datapath_telemetry_record_kafka_to_clickhouse # [string] Job Name
+    shade.identifier: aes
+    pipeline:
+      object-reuse: true # [boolean] Object Reuse, default is false
+  topology:
+  - name: kafka_source
+    downstream: [etl_processor]
+  - name: etl_processor
+    downstream: [clickhouse_sink]
+  - name: clickhouse_sink
+
+