Apache Beam实战指南 | 手把手教你玩转KafkaIO与Flink

// Define the schema for the records.
Schema appSchema = 
  Schema
    .builder()
    .addInt32Field("appId")
    .addStringField("description")
    .addDateTimeField("rowtime")
    .build();
// Create a concrete row with that type.
Row row = 
  Row
    .withSchema(appSchema)
    .addValues(1, "Some cool app", new Date())
    .build();
// Create a source PCollection containing only that row
PCollection<Row> testApps = 
  PBegin
    .in(p)
    .apply(Create
              .of(row)
              .withCoder(appSchema.getRowCoder()));

// An example POJO class.
class AppPojo {
 Integer appId;
 String description;
 Date timestamp;
}
// Acquire a collection of POJOs somehow.
 PCollection<AppPojo> pojos = ...
// Convert them to Rows with the same schema as defined above via a DoFn.
PCollection<Row> apps = pojos
 .apply(
     ParDo.of(new DoFn<AppPojo, Row>() {
       @ProcessElement
       public void processElement(ProcessContext c) {
         // Get the current POJO instance
         AppPojo pojo = c.element();
         // Create a Row with the appSchema schema 
           // and values from the current POJO
           Row appRow = 
                 Row
                   .withSchema(appSchema)
                  .addValues(
                     pojo.appId, 
                     pojo.description, 
                    pojo.timestamp)
                   .build();
         // Output the Row representing the current POJO
         c.output(appRow);
       }
     }));

* <h3>Supported Kafka Client Versions</h3>
 * KafkaIO relies on <i>kafka-clients</i> for all its interactions with the Kafka cluster.
 * <i>kafka-clients</i> versions 0.10.1 and newer are supported at runtime. The older versions
 * 0.9.x - 0.10.0.0 are also supported, but are deprecated and likely be removed in near future.
 * Please ensure that the version included with the application is compatible with the version of
 * your Kafka cluster. Kafka client usually fails to initialize with a clear error message in
 * case of incompatibility.
 */

<dependency>
   <groupId>org.apache.beam</groupId>
   <artifactId>beam-sdks-java-io-kafka</artifactId>
   <version>...</version>
</dependency>
<dependency>
 <groupId>org.apache.kafka</groupId>
 <artifactId>kafka-clients</artifactId>
 <version>a_recent_version</version>
 <scope>runtime</scope>
</dependency>

pipeline.apply(KafkaIO.<Long, String>read() 
       .withBootstrapServers("broker_1:9092,broker_2:9092")// 
       .withTopic("my_topic")    // use withTopics(List<String>) to read from multiple topics. 
       .withKeyDeserializer(LongDeserializer.class) 
      .withValueDeserializer(StringDeserializer.class) 
      // Above four are required configuration. returns   PCollection<KafkaRecord<Long, String>> 
      // Rest of the settings are optional :  
       // you can further customize KafkaConsumer used to read the records by   adding more 
      // settings for ConsumerConfig. e.g : 
      .updateConsumerProperties(ImmutableMap.of("group.id",   "my_beam_app_1")) 

       // set event times and watermark based on 'LogAppendTime'. To provide   a custom 
      // policy see withTimestampPolicyFactory(). withProcessingTime() is   the default. 
      // Use withCreateTime() with topics that have 'CreateTime' timestamps. 
       .withLogAppendTime()  
      // restrict reader to committed messages on Kafka (see method   documentation). 
       .withReadCommitted()  
       // offset consumed by the pipeline can be committed back. 
       .commitOffsetsInFinalize()  
       // finally, if you don't need Kafka metadata, you can drop it.g 
      .withoutMetadata() // PCollection<KV<Long, String>> 
    ) 
    .apply(Values.<String>create()) // PCollection<String>

pipeline.apply(KafkaIO.<Long, String>read()

.withBootstrapServers("broker_1:9092,broker_2:9092")

.withTopic("my_topic") // use withTopics(List<String>) to read from multiple topics.

.withKeyDeserializer(LongDeserializer.class) 
.withValueDeserializer(StringDeserializer.class)

Properties props = new Properties();
props.put("key.deserializer","org.apache.kafka.common.serialization.ByteArrayDeserializer");
props.put("value.deserializer","org.apache.kafka.common.serialization.ByteArrayDeserializer");

.updateConsumerProperties(ImmutableMap.of("group.id", my_beam_app_1"))

.withLogAppendTime()

.withReadCommitted()

set CommitOffsetsInFinalizeEnabled(boolean commitOffsetInFinalize)
.commitOffsetsInFinalize()

.withoutMetadata() // PCollection<KV<Long, String>>

.apply(Values.<String>create()) // PCollection<String>

PCollection<KV<Long, String>> kvColl = ...;
kvColl.apply(KafkaIO.<Long, String>write()
     .withBootstrapServers("broker_1:9092,broker_2:9092")
     .withTopic("results")
     .withKeySerializer(LongSerializer.class)
     .withValueSerializer(StringSerializer.class)
     // You can further customize KafkaProducer used to write the records by adding more
     // settings for ProducerConfig. e.g, to enable compression :
     .updateProducerProperties(ImmutableMap.of("compression.type", "gzip"))
    // You set publish timestamp for the Kafka records.
     .withInputTimestamp() // element timestamp is used while publishing to Kafka
     // or you can also set a custom timestamp with a function.
     .withPublishTimestampFunction((elem, elemTs) -> ...)
     // Optionally enable exactly-once sink (on supported runners). See JavaDoc for withEOS().
     .withEOS(20, "eos-sink-group-id");
  );

.withEOS(20, "eos-sink-group-id");

/** Provided options. */
private final FlinkPipelineOptions options;

public interface FlinkPipelineOptions  extends 
PipelineOptions, ApplicationNameOptions, StreamingOptions {
 //....
}

/**
* The url of the Flink JobManager on which to execute pipelines.   This can either be the the   * address of a cluster JobManager, in the form "host:port" or one of the special Strings  * "[collection]" will execute the pipeline on Java Collections while "[auto]" will let the system
*/
@Description( "Address of the Flink Master where the Pipeline should  be executed. Can"+ "[collection] or [auto].")
void setFlinkMaster(String value);

@Description("The degree of parallelism to be used when distributing operations onto workers.")
@Default.InstanceFactory(DefaultParallelismFactory.class)
Integer getParallelism();
void setParallelism(Integer value);

@Description("The interval between consecutive checkpoints (i.e.  snapshots of the current"
@Default.Long(-1L)
Long getCheckpointingInterval();
void setCheckpointingInterval(Long interval)

@Description("The checkpointing mode that defines consistency guarantee.")
@Default.Enum("AT_LEAST_ONCE")
CheckpointingMode getCheckpointingMode();
void setCheckpointingMode(CheckpointingMode mode);

@Description("The maximum time that a checkpoint may take before being discarded.")
@Default.Long(20 * 60 * 1000)
Long getCheckpointTimeoutMillis();
void setCheckpointTimeoutMillis(Long checkpointTimeoutMillis);

@Description(
"Sets the number of times that failed tasks are re-executed. "
+ "A value of zero effectively disables fault tolerance. A value of -1 indicates "+ "that the system default value (as defined in the configuration) should be used.")
@Default.Integer(-1)
Integer getNumberOfExecutionRetries();
void setNumberOfExecutionRetries(Integer retries);

@Description(
     "Sets the delay between executions. A value of {@code -1} "
         + "indicates that the default value should be used.")
@Default.Long(-1L)
Long getExecutionRetryDelay();
void setExecutionRetryDelay(Long delay);

@Description("Sets the behavior of reusing objects.")
@Default.Boolean(false)
Boolean getObjectReuse();
void setObjectReuse(Boolean reuse);

@Description("Sets the state backend to use in streaming mode. "
@JsonIgnore
AbstractStateBackend getStateBackend();
void setStateBackend(AbstractStateBackend stateBackend);

@Description("Enable/disable Beam metrics in Flink Runner")
@Default.Boolean(true)
BooleangetEnableMetrics();
voidsetEnableMetrics(BooleanenableMetrics);

@Description(
"Enables or disables externalized checkpoints."
+"Works in conjunction with CheckpointingInterval")
@Default.Boolean(false)
BooleanisExternalizedCheckpointsEnabled();
voidsetExternalizedCheckpointsEnabled(BooleanexternalCheckpoints);

@Description("If set, shutdown sources when their watermark reaches +Inf.")
@Default.Boolean(false)
BooleanisShutdownSourcesOnFinalWatermark();
voidsetShutdownSourcesOnFinalWatermark(BooleanshutdownOnFinalWatermark);

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-sdks-java-io-kafka</artifactId>
  <version>2.4.0</version>
 </dependency>
 <dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>0.10.1.1</version>
 </dependency>
 <dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-runners-core-java</artifactId>
  <version>2.4.0</version>
 </dependency>
 <dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-runners-flink_2.11</artifactId>
  <version>2.4.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, 
// 必需设置 kafka 的服务器地址和端口
String>read().withBootstrapServers("192.168.1.110:11092,192.168.1.119:11092,192.168.1.120:11092")
     .withTopic("testmsg")// 必需设置要读取的 kafka 的 topic 名称
     .withKeyDeserializer(StringDeserializer.class)// 必需序列化 key
     .withValueDeserializer(StringDeserializer.class)// 必需序列化 value
     .updateConsumerProperties(ImmutableMap.<String, Object>of("auto.offset.reset", "earliest")));// 这个属性 kafka 最常见的.
 // 为输出的消息类型。或者进行处理后返回的消息类型
PCollection<String> kafkadata = lines.apply("Remove Kafka Metadata", ParDo.of(new DoFn<KafkaRecord<String, String>, String>() { 
private static final long serialVersionUID = 1L;
  @ProcessElement
  public void processElement(ProcessContext ctx) {
   System.out.print("输出的分区为 ----：" + ctx.element().getKV());
   ctx.output(ctx.element().getKV().getValue());// 其实我们这里是把"张海     涛在发送消息 ***"进行返回操作
  }
 }));
PCollection<String> windowedEvents = kafkadata.apply(Window.<String>into(FixedWindows.of(Duration.standardSeconds(5))));
PCollection<KV<String, Long>> wordcount = windowedEvents.apply(Count.<String>perElement()); // 统计每一个 kafka 消息的 Count
PCollection<String> wordtj = wordcount.apply("ConcatResultKVs", MapElements.via( // 拼接最后的格式化输出（Key 为 Word，Value 为 Count）
 new SimpleFunction<KV<String, Long>, String>() {
  private static final long serialVersionUID = 1L;
   @Override
   public String apply(KV<String, Long> input) {
   System.out.print("进行统计：" + input.getKey() + ": " + input.getValue());
     return input.getKey() + ": " + input.getValue();
    }
   }));
 wordtj.apply(KafkaIO.<Void, String>write()  .withBootstrapServers("192.168.1.110:11092,192.168.1.119:11092,192.168.1.120:11092")// 设置写会 kafka 的集群配置地址
   .withTopic("senkafkamsg")// 设置返回 kafka 的消息主题
   // .withKeySerializer(StringSerializer.class)// 这里不用设置了，因为上面 Void 
   .withValueSerializer(StringSerializer.class)
   // Dataflow runner and Spark 兼容， Flink 对 kafka0.11 才支持。我的版本是 0.10 不兼容
   //.withEOS(20, "eos-sink-group-id")
   .values() // 只需要在此写入默认的 key 就行了，默认为 null 值
 ); // 输出结果
 pipeline.run().waitUntilFinish();
｝

//FlinkPipelineOptions options =PipelineOptionsFactory.as(FlinkPipelineOptions.class);
//options.setStreaming(true);
//options.setAppName("app_test");
//options.setJobName("flinkjob");
//options.setFlinkMaster("localhost:6123");
//options.setParallelism(10);// 设置 flink 的并行度
// 显式指定 PipelineRunner：FlinkRunner，必须指定，如果不指定则为本地 
options.setRunner(FlinkRunner.class);

.updateConsumerProperties(ImmutableMap.<String,Object>of("auto.offset.reset", "earliest")));

KafkaIO.<Void, String>write()
.values() // 只需要在此写入默认的 key 就行了，默认为 null 值