flink使用03-数据输入的几种不同方法

flink的数据输入源主要分为两大类:

1. 内置数据源

集合数据源

可以将数组或者集合作为 flink 的数据源,分别有不同的方法可以使用, 这种方式比较适合本地调试使用

// 添加数组作为数据输入源
String[] elementInput = new String[]{"hello Flink", "Second Line"};
DataStream<String> text = env.fromElements(elementInput);

// 添加List集合作为数据输入源
List<String> collectionInput = new ArrayList<>();
collectionInput.add("hello Flink");
DataStream<String> text2 = env.fromCollection(collectionInput);

Socket数据源

// 添加Socket作为数据输入源
// 4个参数 -> (hostname:Ip地址, port:端口, delimiter:分隔符, maxRetry:最大重试次数)
DataStream<String> text3 = env.socketTextStream("localhost", 9999, "\n", 4);

文件数据源

可以使用 readTextFile 方法直接读取文本文件, 这种方式可以用来监控一下 log 日志文件, 也可以使用 readFile 方法通过指定 InputFormat 来读取特定数据类型的文件, InputFormat可以是内置类,如 CsvInputFormat 或者用户自定义 InputFormat 接口类.
```
// 添加文件源
// 直接读取文本文件
DataStream<String> text4 = env.readTextFile("/opt/history.log");

// 指定 CsvInputFormat, 监控csv文件(两种模式), 时间间隔是10ms
        DataStream<String> text5 = env.readFile(new CsvInputFormat<String>(new Path("/opt/history.csv")) {
            @Override
            protected String fillRecord(String s, Object[] objects) {
                return null;
            }
        },"/opt/history.csv", FileProcessingMode.PROCESS_CONTINUOUSLY,10);
```
在 readFile() 方法中有一项参数为 WatchType, 共有两种模式 (PROCESS_CONTINUOUSLY / PROCESS_ONCE). 在 PROCESS_CONTINUOUSLY 模式下, 检测到文件变动就会将文件全部内容加载在 flink, 在 PROCESS_ONCE 模式下, 只会将文件变动的那部分加载到 flink.

2. 外部数据源

外部数据源是重头戏, 一般来说项目中均是使用外部数据源作为数据的源头, flink 通过实现 SourceFunction 定义了非常丰富的第三方数据连接器

数据源连接器

对于第三方数据源, flink的支持分为三种,有只读型(Twitter Streaming API / Netty ), 只写型( Cassandra / Elasticsearch / hadoop FileSystem), 支持读写(Kafka / Amazon Kinesis / RabbitMQ)

Apache Kafka (Source / Sink)

Apache Cassandra (Sink)

Amazon Kinesis Streams (Source / Sink)

Elasticsearch (Sink)

Hadoop FileSystem (Sink)

RabbitMQ (Source / Sink)

Apache NiFI (Source / Sink)

Twitter Streaming API (Source)

Apache Bahir 中的连接器:

Apache ActiveMQ (Source / Sink)

Apache Flume (Sink)

Redis (Sink)

Akka (Sink)

Netty (Source)

以Kafka 为例做演示

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 配置 kafka 连接参数
String topic = "topic_name";
String bootStrapServers = "localhost:9092";
String zkConnect = "localhost:2181";
String groupID = "group_A";
Properties prop = new Properties();
prop.setProperty("bootstrap.servers", bootStrapServers);
prop.setProperty("zookeeper.connect", zkConnect);
prop.setProperty("group.id", groupID);

// 创建 kafka connector source
FlinkKafkaConsumer010<String> consumer010 = new FlinkKafkaConsumer010<>(topic, new SimpleStringSchema(), prop);

// add source
DataStreamSource<String> dataStream = env.addSource(consumer010);

自定义数据源连接器

用户也可以自己定义连接器, 通过实现 SourceFunction 定义单个线程的接入的数据连接器, 也可以通过实现ParallelSourceFunction 接口或者继承 RichParallelSourceFunction 类定义并发数据源接入器.

flink使用03-数据输入的几种不同方法

flink的数据输入源主要分为两大类:

1. 内置数据源

2. 外部数据源

推荐阅读更多精彩内容