从0到1Flink的成长之路

Posted 2021-07-25 熊老二-

数据终端 Sink

API

1.ds.print 直接输出到控制台
2.ds.printToErr() 直接输出到控制台,用红色
3.ds.collect 将分布式数据收集为本地集合
4.ds.setParallelism(1).writeAsText("本地/HDFS的path",WriteMode.OVERWRITE)

注意

在输出到path的时候，可以在前面设置并行度，如果
并行度>1，则path为目录
并行度=1，则path为文件名

代码演示

package xx.xxxxx.flink.sink;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FileSystem;
import java.util.List;
/**
* DataSet API 批处理中数据终端：基于文件Sink
* 1.ds.print 直接输出到控制台
* 2.ds.printToErr() 直接输出到控制台,用红色
* 3.ds.collect 将分布式数据收集为本地集合
* 4.ds.setParallelism(1).writeAsText("本地/HDFS的path",WriteMode.OVERWRITE)
*
* 注意: 在输出到path的时候，可以在前面设置并行度，如果
* 并行度>1，则path为目录
* 并行度=1，则path为文件名
*/
public class BatchSinkFileDemo {
public static void main(String[] args) throws Exception {
// 1. 执行环境-env
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
// 2. 数据源-source：基于文件数据Source
DataSource<String> dataSet = env.fromElements("mapreduce", "spark", "flink");
// 3. 数据终端-sink
dataSet.print();
dataSet.printToErr();
// DataSet转换为本地集合，比如列表List
List<String> list = dataSet.collect();
System.out.println(list);
// 保存数据至文本文件
dataSet.writeAsText("datas/sink.txt").setParallelism(1); // 并行度为1，写入文件
// 保存至CSV文件，数据类型必须为二元组
dataSet
// TODO: 使用加强版映射函数：RichMapFunction
.map(new RichMapFunction<String, Tuple2<Integer, String>>() {
@Override
public Tuple2<Integer, String> map(String value) throws Exception {
// 获取分区索引
int index = getRuntimeContext().getIndexOfThisSubtask();
return new Tuple2<>(index, value);
}
})
.writeAsCsv("datas/sink-csv.txt", FileSystem.WriteMode.OVERWRITE)
.setParallelism(1) ;
// 4.执行-execute
env.execute(BatchSinkFileDemo.class.getSimpleName());
}
}