flink集成MySQL数据源_Flink配置Kafka数据源

简介

正文

1. FlinkKafkaConsumer010

flink中已经预置了kafka相关的数据源实现FlinkKafkaConsumer010&＃xff0c;先看下具体的实现&＃xff1a;

&＃64;PublicEvolving

public class FlinkKafkaConsumer010 extends FlinkKafkaConsumer09 {

private static final long serialVersionUID &＃61; 2324564345203409112L;

public FlinkKafkaConsumer010(String topic, DeserializationSchema valueDeserializer, Properties props) {

this(Collections.singletonList(topic), valueDeserializer, props);

}

public FlinkKafkaConsumer010(String topic, KeyedDeserializationSchema deserializer, Properties props) {

this(Collections.singletonList(topic), deserializer, props);

}

public FlinkKafkaConsumer010(List topics, DeserializationSchema deserializer, Properties props) {

this((List)topics, (KeyedDeserializationSchema)(new KeyedDeserializationSchemaWrapper(deserializer)), props);

}

public FlinkKafkaConsumer010(List topics, KeyedDeserializationSchema deserializer, Properties props) {

super(topics, deserializer, props);

}

&＃64;PublicEvolving

public FlinkKafkaConsumer010(Pattern subscriptionPattern, DeserializationSchema valueDeserializer, Properties props) {

this((Pattern)subscriptionPattern, (KeyedDeserializationSchema)(new KeyedDeserializationSchemaWrapper(valueDeserializer)), props);

}

&＃64;PublicEvolving

public FlinkKafkaConsumer010(Pattern subscriptionPattern, KeyedDeserializationSchema deserializer, Properties props) {

super(subscriptionPattern, deserializer, props);

}

......

}

kafka的Consumer有一堆实现&＃xff0c;不过最终都是继承自FlinkKafkaConsumerBase&＃xff0c;而这个抽象类则是继承RichParallelSourceFunction&＃xff0c;是不是很眼熟&＃xff0c;跟自定义mysql数据源继承的抽象类RichSourceFunction很类似。

public abstract class FlinkKafkaConsumerBase

extends RichParallelSourceFunction

implements CheckpointListener, ResultTypeQueryable, CheckpointedFunction

可以看到&＃xff0c;这里有很多构造函数&＃xff0c;我们直接使用即可。

1.1 代码使用

package myflink.job;

import org.apache.flink.api.common.serialization.SimpleStringSchema;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;

import java.util.Properties;

/**

* kafka作为数据源&＃xff0c;消费kafka中的消息

* 教程详见

* &＃64;See http://www.54tianzhisheng.cn/tags/Flink/

*/

public class KafkaDatasouceForFlinkJob {

public static void main(String[] args) throws Exception {

final StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment();

Properties properties &＃61; new Properties();

properties.put("bootstrap.servers","localhost:9092");

properties.put("zookeeper.connect","localhost:2181");

properties.put("group.id","metric-group");

properties.put("auto.offset.reset","latest");

properties.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");

properties.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");

DataStreamSource dataStreamSource &＃61; env.addSource(

new FlinkKafkaConsumer010(

"testjin" ,// topic

new SimpleStringSchema(),

properties

)

).setParallelism(1);

// dataStreamSource.print();

// 同样效果

dataStreamSource.addSink(new PrintSinkFunction<>());

env.execute("Flink add kafka data source");

}

}

说明&＃xff1a;

a、这里直接使用properties对象来设置kafka相关配置&＃xff0c;比如brokers、zk、groupId、序列化、反序列化等。

b、使用FlinkKafkaConsumer010构造函数&＃xff0c;指定topic、properties配置

c、SimpleStringSchema仅针对String类型数据的序列化及反序列化&＃xff0c;如果kafka中消息的内容不是String&＃xff0c;则会报错&＃xff1b;看下SimpleStringSchema的定义&＃xff1a;

public class SimpleStringSchema implements DeserializationSchema, SerializationSchema

d、这里直接把获取到的消息打印出来。