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

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

1. 内置数据源

  • 集合数据源

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

    1
    2
    3
    4
    5
    6
    7
    8
    // 添加数组作为数据输入源
    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数据源

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

    可以使用 readTextFile 方法直接读取文本文件, 这种方式可以用来监控一下 log 日志文件, 也可以使用 readFile 方法通过指定 InputFormat 来读取特定数据类型的文件, InputFormat可以是内置类,如 CsvInputFormat 或者用户自定义 InputFormat 接口类.

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    // 添加文件源
    // 直接读取文本文件
    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 为例 做演示

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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 类定义并发数据源接入器.

0%