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 类定义并发数据源接入器.