玩轉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>

.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.ByteArray Deserializer");

set CommitOffsetsInFinalizeEnabled (boolean commitOffsetInFinalize)

.commitOffsetsInFinalize()

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");

);

/** 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)

BooleanisExternalized CheckpointsEnabled();

voidsetExternalizedCheckpointsEnabled (BooleanexternalCheckpoints);

@Description ("If set, s hutdown sources when their watermark reaches +Inf.")

@Default.Boolean (false)

BooleanisShutdownSources OnFinalWatermark();

voidsetShutdown SourcesOnFinalWatermark (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);

KafkaIO.<Void, String>write()

.values() // 只需要在此寫入預設的key就行了，預設為null值