Flink - 读取 Parquet 文件 By Scala / Java

一.引言

parquet 文件常见与 Flink、Spark、Hive、Streamin、MapReduce 等大数据场景，通过列式存储和元数据存储的方式实现了高效的数据存储与检索，下面介绍 Flink 场景下如何读取 Parquet。Parquet 相关知识可以参考：Spark - 一文搞懂 parquet。

二.Parquet Read By Scala

1.依赖准备与环境初始化

import org.apache.hadoop.fs.FileSystem
import org.apache.flink.formats.parquet.ParquetRowInputFormat
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.parquet.hadoop.ParquetFileReader
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
import org.apache.parquet.schema.Type.Repetition
import org.apache.parquet.schema.{MessageType, PrimitiveType, Type}

Flink 读取 parquet 除了正常 Flink 环境相关依赖外，还需要加载单独的 Parquet 组件：

        <dependency>
            <groupId>org.apache.parquetgroupId>
            <artifactId>parquet-avroartifactId>
            <version>1.10.0version>
        dependency>
 
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-parquet_2.12artifactId>
            <version>1.9.0version>
        dependency>

本文基于 Flink-1.13.1 + scala-2.12.8 + hadoop-2.6.0 的运行环境，不同版本下可能需要更换上述 parquet 相关依赖。下面初始化 Flink ExecutionEnvironment，因为流式处理的原因，这里初始化环境类型为 Stream:

    val env = StreamExecutionEnvironment.getExecutionEnvironment

2.推断 Schem 读取 Parquet

parquet 通过列式存储数据，所以需要 schema 标定每一列的数据类型与名称，与 Spark 类似， Flink 也可以通过 Parquet 文件推断其对应 schema 并读取 Parquet。

  def readParquetWithInferSchema(env: StreamExecutionEnvironment): Unit = {
    val filePath = "./test.parquet"
    val configuration = new org.apache.hadoop.conf.Configuration(true)
    val parquetFileReader = ParquetFileReader.readFooter(configuration, new org.apache.hadoop.fs.Path(filePath))
    val schema: MessageType = parquetFileReader.getFileMetaData.getSchema
    println(s"Schema: $schema")
 
    val hdfs: FileSystem = org.apache.hadoop.fs.FileSystem.get(configuration)
 
    val rowData = env.readFile(new ParquetRowInputFormat(new org.apache.flink.core.fs.Path(filePath), schema), filePath).setParallelism(1)
 
    rowData.map(row => {
      val source = row.getField(1)
      val flag = row.getField(35)
      source + "\t" + flag
    }).setParallelism(1).print()
  }

通过 parquetFileReader 获取元数据 MetaData 并获取 parquet 对应 schema，最终通过 env.readFile 方法指定 InputFormat 为 ParquetRowInputFormat 读取 parquet 文件，先看一下打印出来的 schema 形式：

由于读取的 parquet 为 SparkSession 生成，所以列名采用了 Spark 的默认形式 _c1,_c2 ...

    env.execute("ReadParquet")

调用执行方法运行上述 print demo 打印最终结果。

Tips:

这里的 Row 类型为 org.apache.flink.types.Row 而不再是 org.apache.spark.sql.Row，获取元素的方法也不再是 row.getString 或其他，而是采用 getFiled 传入 position 或者 列名 得到，索引从 0 开始。

java 读写Parquet格式的数据的示例代码

pdf 0星 超过10%的资源 62KB

下载

3.指定 schema 读取 Parquet

除了 infer 推理得到 schema 外，读取也支持自定义 schema，与 spark 类似，这里也提供了 PrimitiveType 指定每一列的数据类型，并合并为 MessageType 得到最终的 schema。

  def readParquetWithAssignSchema(env: StreamExecutionEnvironment): Unit = {
    val filePath = "./test.parquet"
    val id = new PrimitiveType(Repetition.OPTIONAL, PrimitiveTypeName.BINARY, "_c0")
    val source = new PrimitiveType(Repetition.OPTIONAL, PrimitiveTypeName.BINARY, "_c1")
    val flag = new PrimitiveType(Repetition.OPTIONAL, PrimitiveTypeName.BINARY, "_c35")
    val typeArray = Array(id, source, flag)
    val typeListAsJava = java.util.Arrays.asList(typeArray: _*).asInstanceOf[java.util.List[Type]]
 
    val schema = new MessageType("schema", typeListAsJava)
    println(schema)
    val rowData = env.readFile(new ParquetRowInputFormat(new org.apache.flink.core.fs.Path(filePath), schema), filePath).setParallelism(1)
 
    rowData.map(row => {
      val source = row.getField(1)
      val flag = row.getField(2)
      source + "\t" + flag
    }).setParallelism(1).print()
  }

上面读取的 test.parquet 有 40+ col，这里只读取第 1，2，35 列，所以单独指定 id，source，flag 三列生成 PrimitiveType 并添加至 MessageType 形成 schema，由于 MessageType 为 Java 参数，所以需要通过 asList + asInstance 进行转化，看一下当前的 schema 情况:

    env.execute("ReadParquet")

调用执行方法执行上述 print 逻辑即可。

Tips:

这里列名给出了 _c0, _c1,_c35，但是读取是 position 索引只能选取 0,1,2，因为 schema 数量决定了读取 Row 的列数，而 schema 的列名决定了读取的内容，在该 schema 基础下读取 getField(35) 会报数组越界  java.lang.ArrayIndexOutOfBoundsException: 

三. Parquet Read By Java

《从0到1学习Flink》—— Flink 读取 Kafka 数据批量写入到 MySQL.pdf

pdf 1星 超过10%的资源 1.09MB

下载

java 读取与 scala 大同小异，主要差别是 map 变为 MapFunction，这里直接贴完整函数方法：

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.formats.parquet.ParquetRowInputFormat;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.types.Row;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
import org.apache.parquet.schema.MessageType;
 
/**
 * @title: ReadParquetByJava
 * @Author DDD
 * @Date: 2022/7/21 8:36 上午
 * @Version 1.0
 */
public class ReadParquetByJava {
 
    public static void main(String[] args) throws Exception {
 
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
        String path = "./test.parquet";
 
        Configuration configuration = new org.apache.hadoop.conf.Configuration(true);
        FileSystem hdfs = org.apache.hadoop.fs.FileSystem.get(configuration);
 
        ParquetMetadata parquetFileReader = ParquetFileReader.readFooter(configuration, new org.apache.hadoop.fs.Path(path));
        MessageType schema = parquetFileReader.getFileMetaData().getSchema();
        System.out.println("-----Schema-----");
        System.out.println(schema);
 
        env.readFile(new ParquetRowInputFormat(new org.apache.flink.core.fs.Path(path), schema), path)
                .setParallelism(1)
                .map(new MapFunction() {
                    @Override
                    public String map(Row row) throws Exception {
                        try {
                            String source = String.valueOf(row.getField(1));
                            String flag = String.valueOf(row.getField(35));
                            return source + "\t" + flag;
                        } catch (Exception e) {
                            e.printStackTrace();
                            return null;
                        }
                    }
                }).print();
        env.execute("ReadParquetByJava");
    }
 
}

四.总结

Parquet 通过其列式存储与空间压缩应用于多种大数据场景，上面给出了 parquet 文件转 DataStream 的两种方式，同理也可以使用 DataSet 加载为静态数据，上面两个方法都给出了 hdfs: FileSystem 变量但都没有使用，下面说下使用场景：

一般分布式任务读取时对应的 parquet 文件不是一个而是多个，所以需要从目标目录中找出第一个合法的 parquet 文件供 ParquetFileReader 解析对应的 schema，hdfs 的任务就是通过目标路径获取第一个合法文件使用。

  def getFirstFilePath(hdfsPath: String, hdfs: FileSystem): String = {
    val files = hdfs.listFiles(new org.apache.hadoop.fs.Path(hdfsPath), false)
    var flag = true
    var firstFile = ""
    while (flag) {
      if (files.hasNext) {
        firstFile = files.next().getPath.getName
        if (!firstFile.equalsIgnoreCase(s"_SUCCESS")
          && !firstFile.startsWith(".")
          && firstFile.endsWith(".parquet")) {
          flag = false
        }
      } else {
        flag = false
      }
    }
    hdfsPath + "/" + firstFile
  }

合法的判断需要三个条件：

A.不包含 _SUCCESS

B.不以 '.' 开头

C.以 '.parquet' 结尾