Flink1.4 HDFS Connector

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 原文来源于:Flink1.4 HDFS Connector此连接器提供一个 Sink,将分区文件写入 Hadoop FileSystem 支持的任何文件系统。

原文来源于:Flink1.4 HDFS Connector

此连接器提供一个 Sink,将分区文件写入 Hadoop FileSystem 支持的任何文件系统。要使用此连接器,添加以下依赖项:

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-filesystem_2.10</artifactId>
  <version>1.4-SNAPSHOT</version>
</dependency>

备注

streaming 连接器目前还不是二进制发布包的一部分,请参阅此处来了解有关如何将程序与Libraries打包以进行集群执行的信息。

文件分桶的Sink(Bucketing File Sink)

分桶(Bucketing)行为以及写入数据操作都可以配置,我们稍后会讲到。下面展示了如何通过默认配置创建分桶的Sink,输出到按时间切分的滚动文件中:

Java版本:

DataStream<String> input = ...;
input.addSink(new BucketingSink<String>("/base/path"));


Scala版本:

val input: DataStream[String] = ...
input.addSink(new BucketingSink[String]("/base/path"))


这里唯一必需的参数是这些分桶文件存储的基本路径 /base/path。可以通过指定自定义 bucketerwriter 和 batch 大小来进一步配置 sink

默认情况下,分桶 sink 根据元素到达时当前系统时间来进行切分,并使用 yyyy-MM-dd--HH 时间格式来命名这些分桶。这个时间格式传递给当前的系统时间的 SimpleDateFormat 来命名桶的路径。每当遇到一个新的时间就会创建一个新的桶。例如,如果你有一个包含分钟的最细粒度时间格式,那么你将会每分钟获得一个新桶。每个桶本身就是一个包含 part 文件的目录:Sink的每个并行实例都将创建自己的 part 文件,当 part 文件变得太大时,会紧挨着其他文件创建一个新的 part 文件。当一个桶在最近没有被写入数据时被视为非活跃的。当桶变得不活跃时,打开的 part 文件将被刷新(flush)并关闭。默认情况下,sink 每分钟都会检查非活跃的桶,并关闭一分钟内没有写入数据的桶。可以在 BucketingSink上 使用 setInactiveBucketCheckInterval() 和 setInactiveBucketThreshold() 配置这些行为。

你还可以使用 BucketingSink上 的 setBucketer() 指定自定义 bucketer。如果需要,bucketer 可以使用元素或元组的属性来确定 bucket目录。

默认的 writer 是StringWriter。对传入的元素调用 toString(),并将它们写入 part 文件,用换行符分隔。要在 BucketingSink 上指定一个自定义的 writer,使用 setWriter() 方法即可。如果要写入 Hadoop SequenceFiles 文件中,可以使用提供的 SequenceFileWriter,并且可以配置使用压缩格式。

最后一个配置选项是 batch 大小。这指定何时关闭 part 文件,并开启一个新文件。(默认part文件大小为384MB)。

Java版本:

DataStream<Tuple2<IntWritable,Text>> input = ...;

BucketingSink<String> sink = new BucketingSink<String>("/base/path");
sink.setBucketer(new DateTimeBucketer<String>("yyyy-MM-dd--HHmm"));
sink.setWriter(new SequenceFileWriter<IntWritable, Text>());
sink.setBatchSize(1024 * 1024 * 400); // this is 400 MB,

input.addSink(sink);


Scala版本:

val input: DataStream[Tuple2[IntWritable, Text]] = ...

val sink = new BucketingSink[String]("/base/path")
sink.setBucketer(new DateTimeBucketer[String]("yyyy-MM-dd--HHmm"))
sink.setWriter(new SequenceFileWriter[IntWritable, Text]())
sink.setBatchSize(1024 * 1024 * 400) // this is 400 MB,

input.addSink(sink)


上面例子将创建一个sink,写入遵循下面格式的分桶文件中:

/base/path/{date-time}/part-{parallel-task}-{count}


其中 date-time 是从日期/时间格式获得的字符串, parallel-task 是并行 sink 实例的索引,count 是由于 batch大小而创建的part文件的运行编号。

备注:

Sink版本:1.4

原文:https://ci.apache.org/projects/flink/flink-docs-release-1.4/dev/connectors/filesystem_sink.html

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
目录
相关文章
|
6天前
|
Java 关系型数据库 分布式数据库
Flink报错问题之flink-sql写hdfs报错如何解决
Flink报错通常是指在使用Apache Flink进行实时数据处理时遇到的错误和异常情况;本合集致力于收集Flink运行中的报错信息和解决策略,以便开发者及时排查和修复问题,优化Flink作业的稳定性。
|
4月前
|
SQL API 流计算
Flink CDC有用iceberg写hdfs的吗?
Flink CDC有用iceberg写hdfs的吗?
61 25
|
5月前
|
SQL 存储 大数据
flink standalone 部署模式且不能使用 hdfs 场景下的各种问题及其应对方案
flink standalone 部署模式且不能使用 hdfs 场景下的各种问题及其应对方案
|
7月前
|
存储 SQL 分布式计算
阿里云全托管flink-vvp平台hudi connector实践(基于emr集群oss-hdfs存储)
阿里云全托管flink-vvp平台hudi sink connector实践,本文数据湖hudi基于阿里云E-MapReduce产品,以云对象存储oss-hdfs作为存储
|
分布式计算 资源调度 Hadoop
在文件存储HDFS版上使用 Apache Flink
本文档主要介绍如何在挂载文件存储HDFS版的 Hadoop 集群上安装及使用 Flink。
287 0
|
SQL HIVE 流计算
flink写hdfs inprogress文件问题
flink落地到hdfs的文件,在hive中读不到
1849 0
|
域名解析 分布式计算 资源调度
在文件存储 HDFS 上使用 Apache Flink
本文主要为大家介绍在文件存储HDFS上搭建及使用Apache Flink的方法。
1244 0
在文件存储 HDFS 上使用 Apache Flink
|
存储 安全 双11
阿里巴巴大规模应用Flink的踩坑经验:如何大幅降低 HDFS 压力?
众所周知 Flink 是当前广泛使用的计算引擎,Flink 使用 checkpoint 机制进行容错处理[1],Flink 的 checkpoint 会将状态快照备份到分布式存储系统,供后续恢复使用。在 Alibaba 内部我们使用的存储主要是 HDFS,当同一个集群的 Job 到达一定数量后,会对 HDFS 造成非常大的压力,本文将介绍一种大幅度降低 HDFS 压力的方法 -- 小文件合并。
|
流计算 Java SQL
Flink落HDFS数据按事件时间分区解决方案
0x1 摘要 Hive离线数仓中为了查询分析方便,几乎所有表都会划分分区,最为常见的是按天分区,Flink通过以下配置把数据写入HDFS, BucketingSink<Object> sink = new BucketingSink<>(path); //通过这样的方式来实现数据跨天分区 sink.
4235 0
|
10天前
|
消息中间件 Kafka Apache
Apache Flink 是一个开源的分布式流处理框架
Apache Flink 是一个开源的分布式流处理框架
163 5