应用开发

快速创建项目

$ mvn archetype:generate                \
  -DarchetypeGroupId=org.apache.flink   \
  -DarchetypeArtifactId=flink-quickstart-java \
  -DarchetypeVersion=1.14.3

Project Configuration

创建执行环境

// 批处理执行环境
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

// 流处理执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

用parameter tool工具从程序启动参数中提取配置项

ParameterTool parameterTool = ParameterTool.fromArgs(args);
String host = parameterTool.get("host");
int port = parameterTool.getInt("port");

./bin/flink run -c com.atguigu.wc.StreamWordCount -p 3 /path/flink.jar --host localhost --port 7777

读取数据

文本文件

// 从文件中读取数据（批处理）
String inputPath = "/Users/haseochen/Documents/HJ/project/实时数仓/Flink/FlinkTutorial/src/main/resources/hello.txt";
DataSet<String> inputDataSet = env.readTextFile(inputPath);

// 从文件中读取数据（流处理）
String inputPath = "/Users/haseochen/Documents/HJ/project/实时数仓/Flink/FlinkTutorial/src/main/resources/hello.txt";
DataStream<String> inputDataStream = env.readTextFile(inputPath);
// 执行任务
env.execute();

从集合中读取数据

// 从集合中读取数据
DataStream<SensorReading> dataStream = env.fromCollection(Arrays.asList(
        new SensorReading("sensor_1", 1547718199L, 35.8),
        new SensorReading("sensor_6", 1547718201L, 15.4),
        new SensorReading("sensor_7", 1547718202L, 6.7),
        new SensorReading("sensor_10", 1547718205L, 38.1)
));

DataStream<Integer> integerDataStream = env.fromElements(1, 2, 4, 67, 189);

// 打印输出
dataStream.print("data");
integerDataStream.print("int");

读取socket数据

# 监听
nc -lk 7777

// 从socket文本流读取数据
DataStream<String> inputDataStream = env.socketTextStream(host, port);

# 提交任务
./bin/flink run -c com.atguigu.wc.StreamWordCount -p 3 /path/flink.jar --host localhost --port 7777