Apache Beam 实战指南 | 如何结合 ClickHouse 打造“AI 微服务”？

 switch (columnType.typeName()) {      case FLOAT32:        stream.writeFloat32((Float) value);        break;
      case FLOAT64:        stream.writeFloat64((Double) value);        break;     .....

 public static <T> Write<T> write(String jdbcUrl, String table)  {...

ClickHouseIO.<Row>write("jdbc:clickhouse://101.201.*.*:8123/Alarm", "AlarmTable")

public Write<T> withMaxInsertBlockSize(long value) {      return toBuilder().maxInsertBlockSize(value).build();  }

 public Write<T> withMaxRetries(int value) {      return toBuilder().maxRetries(value).build(); }

withInsertDistributedSync(@Nullable Boolean value) {..

 withInsertQuorum（@Nullable  Long  value）{..

 withInsertDeduplicate（Boolean  value）{..

  withMaxCumulativeBackoff（Duration  value）  withInitialBackoff（Duration  value）{

      <!-- 本地运行Runners --><dependency><groupId>org.apache.beam</groupId><artifactId>beam-runners-direct-java</artifactId><version>2.11.0</version></dependency><!-- 运行Runners核心jar --><dependency><groupId>org.apache.beam</groupId><artifactId>beam-runners-core-java</artifactId><version>2.11.0</version></dependency><!-- 引入Beam clickhouse jar --><dependency><groupId>org.apache.beam</groupId><artifactId>beam-sdks-java-io-clickhouse</artifactId><version>2.11.0</version></dependency><dependency><groupId>ru.yandex.clickhouse</groupId><artifactId>clickhouse-jdbc</artifactId><version>0.1.47</version></dependency><dependency><groupId>org.apache.beam</groupId><artifactId>beam-sdks-java-core</artifactId><version>2.11.0</version></dependency>   <!-- BeamSQL --><dependency><groupId>org.apache.beam</groupId><artifactId>beam-sdks-java-extensions-sql</artifactId><version>2.11.0</version></dependency>       <!--引入elasticsearch--><dependency><groupId>org.apache.beam</groupId><artifactId>beam-sdks-java-io-elasticsearch</artifactId><version>2.11.0</version></dependency><!-- kafka --><dependency><groupId>org.apache.beam</groupId><artifactId>beam-sdks-java-io-kafka</artifactId><version>2.11.0</version></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.0.0</version></dependency><!-- Flink --><dependency><groupId>org.apache.beam</groupId><artifactId>beam-runners-flink_2.11</artifactId><version>2.11.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.5.2</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_2.11</artifactId><version>1.5.2</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-core</artifactId><version>1.5.2</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-runtime_2.11</artifactId><version>1.5.2</version><!--<scope>provided</scope> --></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_2.11</artifactId><version>1.5.2</version><!--<scope>provided</scope> --></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-metrics-core</artifactId><version>1.5.2</version><!--<scope>provided</scope> --></dependency>

public static void main(String[] args) {
// 创建管道工厂PipelineOptions options = PipelineOptionsFactory.create();// 显式指定PipelineRunner：FlinkRunner必须指定如果不制定则为本地options.setRunner(FlinkRunner.class);Pipeline pipeline = Pipeline.create(options);// 设置相关管道// 这里kV后说明kafka中的key和value均为String类型PCollection<KafkaRecord<String, String>> lines =  pipeline.apply(KafkaIO.<String,String>read().withBootstrapServers("101.*.*.77:9092")// 必需设置kafka的服务器地址和端口.withTopic("TopicAlarm")// 必需，设置要读取的kafka的topic名称 .withKeyDeserializer(StringDeserializer.class)// 必需序列化key .withValueDeserializer(StringDeserializer.class)// 必需序列化value.updateConsumerProperties(ImmutableMap.<String, Object>of("auto.offset.reset", "earliest")));// 这个属性kafka最常见的//设置Schema 的的字段名称和类型final Schema type = Schema.of(Schema.Field.of("alarmid", FieldType.STRING), Schema.Field.of("alarmTitle", FieldType.STRING),Schema.Field.of("deviceModel", FieldType.STRING), Schema.Field.of("alarmSource", FieldType.INT32), Schema.Field.of("alarmMsg", FieldType.STRING));        //从kafka中读出的数据转换成AlarmTable实体对象PCollection<AlarmTable> kafkadata = lines.apply("Remove Kafka Metadata", ParDo.of(new DoFn<KafkaRecord<String, String>, AlarmTable>() {private static final long serialVersionUID = 1L;@ProcessElementpublic void processElement(ProcessContext ctx) {Gson gon = new Gson();AlarmTable modelTable = null;try {//进行序列号代码modelTable = gon.fromJson(ctx.element().getKV().getValue(),AlarmTable.class);} catch (Exception e) {System.out.print("json序列化出现问题：" + e);}ctx.output(modelTable);//回传实体}}));//备份写入ElasticsearchString[] addresses = { "http://101.*.*.77:9200/" };PCollection<String> jsonCollection=kafkadata.setCoder(AvroCoder.of(AlarmTable.class)).apply("covert json", ParDo.of(new DoFn<AlarmTable, String>() {private static final long serialVersionUID = 1L;@ProcessElementpublic void processElement(ProcessContext ctx) {Gson gon = new Gson();String  jString="";try {// 进行序列号代码jString = gon.toJson(ctx.element());System.out.print("序列化后的数据：" + jString);} catch (Exception e) {System.out.print("json序列化出现问题：" + e);}ctx.output(jString);// 回传实体}}));// 所有的Beam 数据写入ES的数据统一转换成json 才可以正常插入jsonCollection.apply( ElasticsearchIO.write().withConnectionConfiguration(ElasticsearchIO.ConnectionConfiguration.create(addresses, "alarm", "TopicAlarm")));
PCollection<Row> modelPCollection = kafkadata//.setCoder(AvroCoder.of(AlarmTable.class))//如果上面设置下面就不用设置.apply(ParDo.of(new DoFn<AlarmTable, Row>() {//实体转换成Rowprivate static final long serialVersionUID = 1L;@ProcessElement   public void processElement(ProcessContext c) {AlarmTable modelTable = c.element();System.out.print(modelTable.getAlarmMsg());Row alarmRow = Row.withSchema(type)    .addValues(modelTable.getAlarmid(),modelTable.getAlarmTitle(), modelTable.getDeviceModel(), modelTable.getAlarmSource(), modelTable.getAlarmMsg()).build();//实体赋值Row类型c.output(alarmRow);}}));      //写入ClickHousemodelPCollection.setRowSchema(type).apply(ClickHouseIO.<Row>write("jdbc:clickhouse://101.201.56.77:8123/Alarm", "AlarmTable").withMaxRetries(3)// 重试次数.withMaxInsertBlockSize(5) // 添加最大块的大小.withInitialBackoff(Duration.standardSeconds(5)).withInsertDeduplicate(true) // 重复数据是否删除.withInsertDistributedSync(false));pipeline.run().waitUntilFinish();}

.setCoder(AvroCoder.of(AlarmTable.class))

String[] addresses = { "http://101.*.*.77:9200/" };