[大数据]Flink-FileSystem

FileSystem

FileSystem连接器提供了对Flink文件系统抽象所支持的文件系统中的分区文件进行访问。

FileSystem连接器本身包含在Flink中，不需要额外的依赖项。但是Flink支持的文件系统原本的依赖需要确保已添加。

FileSystem连接器需要指定相应的格式(如：CSV、JSON、Avro、Parquet、Orc、Debezium-JSON、Canal-JSON和Raw)，以便从文件系统读写行。

FileSystem连接器允许从本地或分布式文件系统进行读写(Flink支持的抽象文件系统)。

FileSystem连接器支持Source表和Sink表。

FileSystem连接器支持读取文件或者目录。

FileSystem表可定义格式为:

CREATE TABLE MyUserTable (
  column_name1 INT,
  column_name2 STRING,
  ...
  part_name1 INT,
  part_name2 STRING
) PARTITIONED BY (part_name1, part_name2) 
WITH (
  'connector' = 'filesystem',           -- required: 指定连接器
  'path' = 'file:///path/to/whatever',  -- required: 指定文件或者目录
  'format' = '...',                     -- required: 指定表格式
  'partition.default-name' = '...',     -- optional: 如果是动态分区，则为默认分区名——列值为空字符串
  'sink.shuffle-by-partition.enable' = '...', -- 该选项使数据在sink阶段通过动态分区字段Shuffle，这样可以大大减少文件系统接收的文件数量，但可能导致数据倾斜，默认值为false。
  ...
)

1 分区文件

FileSystem连接器分区支持使用标准的Hive分区格式。

不要求分区预先注册到Catalog中，而是根据目录结构发现和推断分区。

例如，根据下面的目录分区的表可以推断出包含日期时间和小时分区：

path
└── datetime=2019-08-25
    └── hour=11
        ├── part-0.parquet
        ├── part-1.parquet
    └── hour=12
        ├── part-0.parquet
└── datetime=2019-08-26
    └── hour=6
        ├── part-0.parquet

FileSystem表同时支持分区插入和覆盖插入。当您向一个分区表插入覆盖时，只有相应的分区会被覆盖，而不是整个表。

2 文件系统源表

FileSystem连接器可用于将单个文件或整个目录读入单个表。

当使用目录作为源路径时，目录内文件的读取顺序不定。

Flink 1.14版本中，FlieSystem流式Source Table仍在开发中。未来，社区将增加对常见流用例的支持，例如分区和目录实时监控。

pom.xml中添加依赖

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-csv</artifactId>
  <version>1.14.3</version>
</dependency>

在本地创建一个csv文件，然后使用Flink SQL创建一个表去映射该数据文件。

package com.qianfeng.day06

import org.apache.flink.table.api._

/**
 * 文件系统数据的获取
 */
object Demo05_SQL_FileSystemSourceTable {
  def main(args: Array[String]): Unit = {
    //表执行环境
    val settings = EnvironmentSettings
      .newInstance()
      .inStreamingMode() // 声明为流任务
      .build()
    val tenv = TableEnvironment.create(settings)
    //创建source表   --  socket
    tenv.executeSql(
      """
        |CREATE TABLE fsTableSource (
        |                  item STRING,
        |                  price DOUBLE,
        |                  proctime as PROCTIME()  -- 该列是机器时间，不需要在数据文件中存在
        |            ) WITH (
        |                'connector' = 'filesystem',
        |                'path' = '/Users/liyadong/data/flinkdata/sql/test.csv',  -- csv文件有两列数据：item和price
        |                'format' = 'csv'
        |           )
        |""".stripMargin)

    //查询表
    tenv.executeSql(
      """
        |select * from fsTableSource
        |""".stripMargin).print()

  }
}

3 流式文件系统Sink表

FileSystem连接器支持流写入，将记录实时写入文件。

Row-encoded(行编码格式)是CSV和JSON。Bulk-encoded(大容量编码格式)是Parquet, ORC和Avro。

可以直接编写SQL，将流数据插入到未分区的表中。如果是分区表，可以配置分区相关操作。

3.1 回滚策略

分区表中的文件落地的条件如下：

注意：

1. 对于Bulk-encoded格式(parquet、orc、avro)，滚动策略结合Checkpoint间隔(依赖下一个检查点完成)控制回滚文件的大小和数量。
2. 对于行格式(csv、json)，如果不想等待很长一段时间才能观察到文件系统中数据的存在，你可以在连接器属性中设置参数sink.rolling-policy.file-size或sink.rolling-policy.rollover-interval 并且同时在flink-conf.yaml文件中设置execute .checkpointing.interval(IDEA在代码中设置也可以)参数。对于其他格式(avro, orc)，你可以在flink-conf.yaml中设置参数execute .checkpointing.interval。

3.2 合并策略

文件接收支持文件合并，即允许应用程序拥有较小的检查点间隔，而无需生成较大文件。

如果启用，文件合并将根据目标文件大小将多个小文件合并为更大的文件。在生产环境中运行文件合并时，请注意:

• 只对单个检查点中的文件进行合并，即生成的文件数量至少与检查点数量相同。
• 合并前的文件是不可见的，因此文件的可见性可能是:检查点间隔+合并时间。
• 如果合并时间过长，会对作业造成压力，并将延长检查点的时间周期。

3.3 分区提交

在写入一个分区之后，通常需要通知下游应用程序。例如，将分区加入到Hive metastore，或者在该目录下写一个_SUCCESS文件。文件系统接收器包含分区提交特性为允许配置自定义策略。提交操作基于触发器和策略的组合。

分区提交触发器(Partition commit trigger):分区提交的时间可以从分区中提取时间的水印或处理时间决定。

分区提交策略(Partition commit policy):如何提交分区，内置策略支持提交成功文件和metastore，也可以实现自己的策略，比如触发hive的生成数据分析统计报告，或者合并小文件等。

==注意:==分区提交只在动态分区插入时起作用。

3.4 分区提交触发器

如果你想让下游尽快看到分区，不管它的数据是否完整:

• ‘sink.partition-commit.trigger’=‘process-time’ (Default value)
• ‘sink.partition-commit.delay’=‘0s’ (Default value) 一旦分区中有数据，它将立即提交。==注意:==分区可能会被提交多次。

如果你想让下游只在数据完成时才能看到这个分区，并且job已经生成了水印和可以从分区值中提取时间:

• ‘sink.partition-commit.trigger’=‘partition-time’
• ‘sink.partition-commit.delay’=‘1h’ (‘1h’ 取决于分区类型)这是提交分区最准确的方式，它会尽量确保提交的分区是尽可能完整的数据。

如果希望下游只在数据完成时才能看到分区，但没有水印，或无法从分区值中提取时间，则:

• ‘sink.partition-commit.trigger’=‘process-time’ (Default value)
• ‘sink.partition-commit.delay’=‘1h’ (‘1h’ 取决于分区类型)尝试准确地提交分区，但数据延迟或故障转移将导致分区过早提交。

延迟数据处理：将延迟数据写入一个已经提交的分区，然后再次触发提交该分区。

3.5 分区时间提取器

默认提取器基于由分区字段组成的时间戳模式。您还可以指定基于PartitionTimeExtractor接口实现的完全自定义分区提取器。

public class HourPartTimeExtractor implements PartitionTimeExtractor {
    @Override
    public LocalDateTime extract(List<String> keys, List<String> values) {
        String dt = values.get(0);
        String hour = values.get(1);
		return Timestamp.valueOf(dt + " " + hour + ":00:00").toLocalDateTime();
	}
}

3.6 分区提交策略

分区提交策略定义了提交分区时采取的操作。

• 第一种是metastore，只有Hive表支持metastore策略，文件系统通过目录结构管理分区。
• 第二个是success文件，它将在分区对应的目录中写入一个空文件。

可以扩展commit policy的实现，自定义commit policy实现如下:

public class AnalysisCommitPolicy implements PartitionCommitPolicy {
    private HiveShell hiveShell;
	
    @Override
	public void commit(Context context) throws Exception {
	    if (hiveShell == null) {
	        hiveShell = createHiveShell(context.catalogName());
	    }
	    
        hiveShell.execute(String.format(
            "ALTER TABLE %s ADD IF NOT EXISTS PARTITION (%s = '%s') location '%s'",
	        context.tableName(),
	        context.partitionKeys().get(0),
	        context.partitionValues().get(0),
	        context.partitionPath()));
	    hiveShell.execute(String.format(
	        "ANALYZE TABLE %s PARTITION (%s = '%s') COMPUTE STATISTICS FOR COLUMNS",
	        context.tableName(),
	        context.partitionKeys().get(0),
	        context.partitionValues().get(0)));
	}
}