Flink流数据采集笔记(三) : Source读取数据编程_org.apache.flink.connector.kafka.source.kafkasourc

目录

一  Flink核心编程概述

二  Environment环境

三  Source

(一)  准备

(二)  从Java中读取数据

(三)  从文件中读取数据

注意事项

(四)  从Socket读取数据

(五)  从Kafka读取数据

(六)  自定义Source 

关键点:

---

一  Flink核心编程概述

        从开发步骤的角度来讲，主要分为四大部分 :

         

二  Environment环境

        没说的,两行代码解决全部

        批:

ExecutionEnvironment benv = ExecutionEnvironment.getExecutionEnvironment();

        流:

  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

三  Source

(一)  准备

        为了更好的方便敲代码,一个导包,一个JavaBean类

<!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <version>1.18.16</version>
</dependency>

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
 
/**
 * 水位传感器：用于接收水位数据
 *
 * id:传感器编号
 * ts:时间戳
 * vc:水位
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
public class WaterSensor {
    private String id;
    private Long ts;
    private Integer vc;
}

(二)  从Java中读取数据

public class SourceJava {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
        env.setParallelism(1);
 
        //2.TODO 从集合中读取数据
        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5);
        DataStreamSource<Integer> streamSource = env.fromCollection(list);
 
        //从元素中读取数据
        DataStreamSource<String> dataStreamSource = env.fromElements("a",