Flink实战(四) - DataSet API编程(下)-阿里云开发者社区

开发者社区> javaedge> 正文

Flink实战(四) - DataSet API编程(下)

简介: Flink实战(四) - DataSet API编程(下)
+关注继续查看

9从压缩文件中创建DataSet

Flink目前支持输入文件的透明解压缩,如果它们标有适当的文件扩展名。 特别是,这意味着不需要进一步配置输入格式,并且任何FileInputFormat都支持压缩,包括自定义输入格式。

压缩文件可能无法并行读取,从而影响作业可伸缩性。

下表列出了当前支持的压缩方法

21.png

9.1 Scala实现

22.png

10 Transformation

10.1 map

Map转换在DataSet的每个元素上应用用户定义的map函数。 它实现了一对一的映射,也就是说,函数必须返回一个元素。


以下代码将Integer对的DataSet转换为Integers的DataSet:


Scala实现

23.png

24.png

Java实现

25.png

10.2 filter

Scala实现

26.png

Java实现

27.png

10.3 mapPartition

MapPartition在单个函数调用中转换并行分区。 map-partition函数将分区作为Iterable获取,并且可以生成任意数量的结果值。 每个分区中的元素数量取决于并行度和先前的操作。

Scala实现

28.png

Java实现

29.png

10.4 first

Scala实现

10.5 Cross

30.png

11 Data Sinks

31.png

32.png

11.1 Java描述

Data Sinks使用DataSet并用于存储或返回它们

使用OutputFormat描述数据接收器操作

Flink带有各种内置输出格式,这些格式封装在DataSet上的操作后面:


writeAsText()/ TextOutputFormat

将元素按行顺序写入字符串。通过调用每个元素的toString()方法获得字符串。

writeAsFormattedText()/ TextOutputFormat

按字符串顺序写入元素。通过为每个元素调用用户定义的format()方法来获取字符串。

writeAsCsv(…)/ CsvOutputFormat

将元组写为逗号分隔值文件。行和字段分隔符是可配置的。每个字段的值来自对象的toString()方法。

print()/ printToErr()/ print(String msg)/ printToErr(String msg)

打印标准输出/标准错误流上每个元素的toString()值。可选地,可以提供前缀(msg),其前缀为输出。这有助于区分不同的打印调用。如果并行度大于1,则输出也将以生成输出的任务的标识符为前缀。

write()/ FileOutputFormat

自定义文件输出的方法和基类。支持自定义对象到字节的转换。

output()/ OutputFormat

最通用的输出方法,用于非基于文件的数据接收器(例如将结果存储在数据库中)。

可以将DataSet输入到多个操作。程序可以编写或打印数据集,同时对它们执行其他转换。

例子

标准数据接收方法:

// text data
DataSet<String> textData = // [...]

// write DataSet to a file on the local file system
textData.writeAsText("file:///my/result/on/localFS");

// write DataSet to a file on a HDFS with a namenode running at nnHost:nnPort
textData.writeAsText("hdfs://nnHost:nnPort/my/result/on/localFS");

// write DataSet to a file and overwrite the file if it exists
textData.writeAsText("file:///my/result/on/localFS", WriteMode.OVERWRITE);

// tuples as lines with pipe as the separator "a|b|c"
DataSet<Tuple3<String, Integer, Double>> values = // [...]
values.writeAsCsv("file:///path/to/the/result/file", "\n", "|");

// this writes tuples in the text formatting "(a, b, c)", rather than as CSV lines
values.writeAsText("file:///path/to/the/result/file");

// this writes values as strings using a user-defined TextFormatter object
values.writeAsFormattedText("file:///path/to/the/result/file",
    new TextFormatter<Tuple2<Integer, Integer>>() {
        public String format (Tuple2<Integer, Integer> value) {
            return value.f1 + " - " + value.f0;
        }
    });

使用自定义输出格式:

DataSet<Tuple3<String, Integer, Double>> myResult = [...]

// write Tuple DataSet to a relational database
myResult.output(
    // build and configure OutputFormat
    JDBCOutputFormat.buildJDBCOutputFormat()
                    .setDrivername("org.apache.derby.jdbc.EmbeddedDriver")
                    .setDBUrl("jdbc:derby:memory:persons")
                    .setQuery("insert into persons (name, age, height) values (?,?,?)")
                    .finish()
    );

本地排序输出

可以使用元组字段位置或字段表达式以指定顺序在指定字段上对数据接收器的输出进行本地排序。 这适用于每种输出格式。

以下示例显示如何使用此功能:

DataSet<Tuple3<Integer, String, Double>> tData = // [...]
DataSet<Tuple2<BookPojo, Double>> pData = // [...]
DataSet<String> sData = // [...]

// sort output on String field in ascending order
tData.sortPartition(1, Order.ASCENDING).print();

// sort output on Double field in descending and Integer field in ascending order
tData.sortPartition(2, Order.DESCENDING).sortPartition(0, Order.ASCENDING).print();

// sort output on the "author" field of nested BookPojo in descending order
pData.sortPartition("f0.author", Order.DESCENDING).writeAsText(...);

// sort output on the full tuple in ascending order
tData.sortPartition("*", Order.ASCENDING).writeAsCsv(...);

// sort atomic type (String) output in descending order
sData.sortPartition("*", Order.DESCENDING).writeAsText(...);

参考

https://nightlies.apache.org/flink/flink-docs-master/docs/dev/dataset/transformations/

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
《HTML5游戏编程核心技术与实战》一2.3 图像API
除了绘制常用的图形以外,canvas提供了一系列的API能够对图像进行操作,常见的图像API有以下3个方法。
1218 0
Flink实战(三) - 编程范式及核心概念(一)
Flink实战(三) - 编程范式及核心概念(一)
5 0
阿里云服务器端口号设置
阿里云服务器初级使用者可能面临的问题之一. 使用tomcat或者其他服务器软件设置端口号后,比如 一些不是默认的, mysql的 3306, mssql的1433,有时候打不开网页, 原因是没有在ecs安全组去设置这个端口号. 解决: 点击ecs下网络和安全下的安全组 在弹出的安全组中,如果没有就新建安全组,然后点击配置规则 最后如上图点击添加...或快速创建.   have fun!  将编程看作是一门艺术,而不单单是个技术。
10318 0
Apache Flink 零基础入门(七):Table API 编程
本文主要包含三部分:第一部分,主要介绍什么是 Table API,从概念角度进行分析,让大家有一个感性的认识;第二部分,从代码的层面介绍怎么使用 Table API;第三部分,介绍 Table API 近期的动态。
1816 0
Kafka实战(六) - 核心API及适用场景全面解析
Kafka实战(六) - 核心API及适用场景全面解析
10 0
+关注
javaedge
关注公众号:JavaEdge,后台回复面试,领取更多大厂求职资源。曾在百度、携程、华为等大厂搬砖,专注Java生态各种中间件原理、框架源码、微服务、中台等架构设计及落地实战,只生产硬核干货!
2317
文章
1
问答
文章排行榜
最热
最新
相关电子书
更多
《2021云上架构与运维峰会演讲合集》
立即下载
《零基础CSS入门教程》
立即下载
《零基础HTML入门教程》
立即下载