Flink读取HDFS上的Parquet文件生成DataSet

你的名字 2021-05-12 11:50 524阅读 0赞

首先打开Flink的官方网站，查看一下DataSet已支持的数据源：  
1、File-Based：

readTextFile(path) / TextInputFormat - Reads files line wise and returns them as Strings.
    
    readTextFileWithValue(path) / TextValueInputFormat - Reads files line wise and returns them as StringValues. StringValues are mutable strings.
    
    readCsvFile(path) / CsvInputFormat - Parses files of comma (or another char) delimited fields. Returns a DataSet of tuples or POJOs. Supports the basic java types and their Value counterparts as field types.
    
    readFileOfPrimitives(path, Class) / PrimitiveInputFormat - Parses files of new-line (or another char sequence) delimited primitive data types such as String or Integer.
    
    readFileOfPrimitives(path, delimiter, Class) / PrimitiveInputFormat - Parses files of new-line (or another char sequence) delimited primitive data types such as String or Integer using the given delimiter.

2、collection-Based：

fromCollection(Collection) - Creates a data set from a Java.util.Collection. All elements in the collection must be of the same type.
    
    fromCollection(Iterator, Class) - Creates a data set from an iterator. The class specifies the data type of the elements returned by the iterator.
    
    fromElements(T ...) - Creates a data set from the given sequence of objects. All objects must be of the same type.
    
    fromParallelCollection(SplittableIterator, Class) - Creates a data set from an iterator, in parallel. The class specifies the data type of the elements returned by the iterator.
    
    generateSequence(from, to) - Generates the sequence of numbers in the given interval, in parallel.

3、Generic：

readFile(inputFormat, path) / FileInputFormat - Accepts a file input format.
    
    createInput(inputFormat) / InputFormat - Accepts a generic input format.

显然，Parquet的读写是第3种--泛型模式，其关键是构建好InputFormat接口的Parquet实现类。Github上clone下flink的源码，IDEA打开并找到InputFormat接口，查看其继承关系。

![format_png][]

RichInputFormat是InputFormat接口的关键实现类，它是一个抽象类，其主要在接口上增加了Flink的运行上下文。

接着打开RichInputFormat抽象类的继承关系：

![format_png 1][]

很明显，FileInputFormat里的ParquetInputFormat就是我们想要的。可以看到，FileInputFormat除了支持Parquet，还支持了Avro、Orc等格式。接着往下跟:

![format_png 2][]

ParquetInputFormat支持Map、Pojo、Row类型，分别对应嵌套、简单实体类、列表，这里选常用的ParquetPojoInputFormat进去看一下：

/**
     * An implementation of {@link ParquetInputFormat} to read POJO records from Parquet files.
     */
    public class ParquetPojoInputFormat<E> extends ParquetInputFormat<E> {
    
        private static final Logger LOG = LoggerFactory.getLogger(ParquetPojoInputFormat.class);
        private final Class<E> pojoTypeClass;
        private final TypeSerializer<E> typeSerializer;
        private transient Field[] pojoFields;
    
        public ParquetPojoInputFormat(Path filePath, MessageType messageType, PojoTypeInfo<E> pojoTypeInfo) {
            super(filePath, messageType);
            this.pojoTypeClass = pojoTypeInfo.getTypeClass();
            this.typeSerializer = pojoTypeInfo.createSerializer(new ExecutionConfig());
            final Map<String, Field> fieldMap = new HashMap<>();
            findAllFields(pojoTypeClass, fieldMap);
            selectFields(fieldMap.keySet().toArray(new String[0]));
        }
        ...
    }

构造函数需要传入3个参数:

*  Path filePath

文件路径，传入路径字符串pathString，然后new Path(pathString)即可。

*  MessageType messageType

Parquet文件的Schema，这个有点难搞，后面我们主要讲这个参数如何构建。

*  PojoTypeInfo<E> pojoTypeInfo

简单实体类的TypeInfo，直接PojoTypeInfo.of(E.class)即可。

要获取Parquet文件的Schema，主要有3种方法:

*  通过Schema字符串构建

/**
     * @description 根据schema字符串创建MessageType
     *
     * @param schema
     * @return org.apache.parquet.schema.MessageType
     */
    public static MessageType build(String schema){
        return MessageTypeParser.parseMessageType(schema);
    }

该种方法需要我们为每个实体类均提前构建好schema字符串，不方便，不用。

*  根据已有parquet文件构建

讲解这种方式前，大家需要了解下Parquet的文件结构:

一个Parquet文件是由一个header以及一个或多个block块组成，以一个footer结尾。header中只包含一个4个字节的数字PAR1用来识别整个Parquet文件格式。文件中所有的metadata都存在于footer中。footer中的metadata包含了格式的版本信息，schema信息、key-value paris以及所有block中的metadata信息。footer中最后两个字段为一个以4个字节长度的footer的metadata,以及同header中包含的一样的PAR1。

读取一个Parquet文件时，需要完全读取Footer的meatadata，Parquet格式文件不需要读取sync markers这样的标记分割查找，因为所有block的边界都存储于footer的metadata中。

所以，只需要读取目标目录中的任何一个Parquet文件即可拿到footer信息，解析footer即可得到Schema。

/**
     * @description 根据Parquet文件创建schema
     *
     * @param parquetFilePath
     * @return org.apache.parquet.schema.MessageType
     */
    public static MessageType buildFromFile(Path parquetFilePath) throws Exception {
        Configuration configuration = new Configuration();
        ParquetMetadata readFooter = null;
        readFooter = ParquetFileReader.readFooter(configuration,
                parquetFilePath, ParquetMetadataConverter.NO_FILTER);
        MessageType schema =readFooter.getFileMetaData().getSchema();
        return schema;
    }
    
    /**
     * @description 根据Parquet文件创建schema
     *
     * @param pathUrl
     * @return org.apache.parquet.schema.MessageType
     */
    public static MessageType buildFromFile(String pathUrl) throws Exception {
        Path parquetFilePath = new Path(pathUrl);
        return buildFromFile(parquetFilePath);
    }

该方法勉强满足我们的需求，但每次读取parquet目录，需要先取一个文件拿来"解剖"拿到Schema才能读取全部文件，听起来就不够优雅，不用。

*  反射

基本思想是通过反射得到实体类各个字段的类型，然后对应成Parquet Schema里的类型。

下面为简单示例：

/**
     * @description 根据类信息动态解析成MessageType
     *
     * @param clazz
     * @param messageName
     * @return org.apache.parquet.schema.MessageType
     */
    public static <T> MessageType build(Class<T> clazz, String messageName){
        Field[] fields = clazz.getDeclaredFields();
        Types.MessageTypeBuilder builder = Types.buildMessage();
        for(Field field: fields){
            field.setAccessible(true);
            String[] fieldTypes = field.getType().toString().toLowerCase().split("\\.");
            String fieldType = fieldTypes[fieldTypes.length-1];
            String fieldName = field.getName();
            if("string".equals(fieldType)){
                builder.required(PrimitiveType.PrimitiveTypeName.BINARY).as(OriginalType.UTF8).named(fieldName);
            }else if("int".equals(fieldType) || "integer".equals(fieldType) || "short".equals(fieldType)){
                builder.required(PrimitiveType.PrimitiveTypeName.INT32).named(fieldName);
            }else if("long".equals(fieldType)){
                builder.required(PrimitiveType.PrimitiveTypeName.INT64).named(fieldName);
            }else if("float".equals(fieldType)){
                builder.required(PrimitiveType.PrimitiveTypeName.FLOAT).named(fieldName);
            }else if("double".equals(fieldType)){
                builder.required(PrimitiveType.PrimitiveTypeName.DOUBLE).named(fieldName);
            }else if("boolean".equals(fieldType)){
                builder.required(PrimitiveType.PrimitiveTypeName.BOOLEAN).named(fieldName);
            }else {
                builder.required(PrimitiveType.PrimitiveTypeName.BINARY).named(fieldName);
            }
        }
        return builder.named(messageName);
    }
    
    /**
     * @description 根据类信息动态解析成MessageType
     *
     * @param clazz
     * @return org.apache.parquet.schema.MessageType
     */
    public static <T> MessageType build(Class<T> clazz){
        String className = clazz.getName();
        return build(clazz, className);
    }

上面的代码勉强可以应付大部分的实体类，但当实体类的字段为Timestamp、Date、List、Map等非基础类型时，则需要继续丰富上述代码，同时还需要考虑字符编码等各类问题才能使上述方法变得健壮。

我，当然可以手动撸一个出来，不怂！！！

但是，太麻烦，肯定有现成的轮子了，何不拿来直接用一下。

回想之前DataStream持久化Parquet文件的时候，曾经用过反射来生成Parquet的Schema，去瞅一眼:

// ParquetAvroWriters.java
    public static <T> ParquetWriterFactory<T> forReflectRecord(Class<T> type) {
        // 基于类反射拿到Avro的Schema
        final String schemaString = ReflectData.get().getSchema(type).toString();
        final ParquetBuilder<T> builder = (out) -> createAvroParquetWriter(schemaString, ReflectData.get(), out);
        return new ParquetWriterFactory<>(builder);
    }

怎么又扯出来Avro了，其实就是用了下Avro的相关转换类来完成Parquet文件Schema的生成。

基本逻辑就是，Avro工具类基于实体类的Class通过反射生成Avro文件Schema，然后Avro工具类再将Avro文件Schema转化成Parquet文件的Schema。

跟踪一下Avro的相关代码，一通乱点，盲猜到上述Avro工具类为AvroSchemaConverter。

点进去看一下其主要方法:

![format_png 3][]

哈哈哈哈哈哈，果然是你，ok，大功告成，调用AvroSchemaConverter工具类的convert方法即可将Avro的Schema转化为Parquet的Schema，即MessageType。

public MessageType convert(Schema avroSchema) {
        if (!avroSchema.getType().equals(Type.RECORD)) {
            throw new IllegalArgumentException("Avro schema must be a record.");
        } else {
            return new MessageType(avroSchema.getFullName(), this.convertFields(avroSchema.getFields()));
        }
    }

封装成一个方法：

public class MessageTypeBuilderByAvro{
        public static <T> MessageType build(Class<T> tClass){
            Schema avroSchema = ReflectData.get().getSchema(tClass);
            MessageType messageType = new AvroSchemaConverter().convert(avroSchema);
            return messageType;
        }
    }

自此，ParquetPojoInputFormat的3个参数均完美构建，下面就可以在Flink中愉快的读取HDFS上的Parquet文件并生成DataSet了，美滋滋！！！

DataSource<E> PojoDataSource = env.createInput(new ParquetPojoInputFormat<E>(
            new Path(pathString),
            MessageTypeBuilderByAvro.build(E.class),
            (PojoTypeInfo<E>) PojoTypeInfo.of(E.class)
        ));

[format_png]: /images/20210511/4629ac7ec19c4f0eab33b2c649e93348.png
[format_png 1]: /images/20210511/933c502291fd48ecb017da8b055538fe.png
[format_png 2]: /images/20210511/aa0a5ebc58b9494c9b4d252030a61465.png
[format_png 3]: /images/20210511/15eccb1add1f4ed8818be6fccb38ccd5.png