Flink之FileSink将数据写入parquet文件
在使用FileSink将数据写入列式存储文件中时必须使用forBulkFormat
,列式存储文件如ORCFile
、ParquetFile
,这里就以ParquetFile
为例结合代码进行说明.
在Flink1.15.3
中是通过构造ParquetWriterFactory
然后调用forBulkFormat
方法将构造好的ParquetWriterFactory
传入,这里先讲一下构造ParquetWriterFactory
一共有三种方式
序列 | API |
---|---|
方式一 | AvroParquetWriters.forGenericRecord |
方式二 | AvroParquetWriters.forSpecificRecord |
方式三 | AvroParquetWriters.forReflectRecord |
其中方式三AvroParquetWriters.forReflectRecord
是我们常用的方法,使用起来也是复杂最低、代码变更时灵活度较好方法,方式二AvroParquetWriters.forSpecificRecord
使用起来复杂度较高,但是代码变更的时候灵活度相对较好的方法,方式一AvroParquetWriters.forGenericRecord
使用起来比较麻烦,而且代码变更时需要更改的也比较多,这里主要介绍方式二和方式三的使用方式.
要说明一点再Flink1.15.3
中是通过AvroParquetWriters
来构造ParquetWriterFactory
,如果是早期版本的Flink可能是要通过ParquetAvroWriters
来进行构造,当然在1.15.3
中也可以通过这个方式进行构造,不过ParquetAvroWriters
已经标注为过时并且建议使用AvroParquetWriters
源码内容如下:
/**
* Convenience builder to create {@link ParquetWriterFactory} instances for the different Avro
* types.
*
* @deprecated use {@link AvroParquetWriters} instead. // 看这部分是建议使用AvroParquetWriters
*/
@Deprecated // 这里已经标注了过时
public class ParquetAvroWriters {
/**
* Creates a ParquetWriterFactory for an Avro specific type. The Parquet writers will use the
* schema of that specific type to build and write the columnar data.
*
* @param type The class of the type to write.
*/
public static <T extends SpecificRecordBase> ParquetWriterFactory<T> forSpecificRecord(
Class<T> type) {
return AvroParquetWriters.forSpecificRecord(type);
}
-
AvroParquetWriters.forReflectRecord(方式三)
这里就先介绍一下
AvroParquetWriters.forReflectRecord
的使用方式,我们在使用FileSink
时最好配合Checkpoint
使用,不然文件只会出现inprogress
状态,感兴趣的可以自己实验一下,我在Flink中FileSink的使用演示了加Checkpoint
和不加Checkpoint
的区别感兴趣的可以看一下.代码模板内容比较简单,直接代码演示:
import com.jin.bean.User; import com.jin.schema.UserSchemaBean; import org.apache.flink.connector.file.sink.FileSink; import org.apache.flink.core.fs.Path; import org.apache.flink.formats.parquet.ParquetWriterFactory; import org.apache.flink.formats.parquet.avro.AvroParquetWriters; import org.apache.flink.streaming.api.CheckpointingMode; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.CheckpointConfig; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner; import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy; /** * @Author: J * @Version: 1.0 * @CreateTime: 2023/6/28 * @Description: 测试 **/ public class FlinkFileSinkForParquet { public static void main(String[] args) throws Exception { // 创建流环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 设置并行度 env.setParallelism(1); // 每30秒作为checkpoint的一个周期 env.enableCheckpointing(30000); // 两次checkpoint间隔最少是20秒 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(20000); // 程序取消或者停止时不删除checkpoint env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // checkpoint必须在60秒结束,否则将丢弃 env.getCheckpointConfig().setCheckpointTimeout(60000); // 同一时间只能有一个checkpoint env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); // 设置EXACTLY_ONCE语义,默认就是这个 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); // checkpoint存储位置 env.getCheckpointConfig().setCheckpointStorage("file:///Users/xxx/data/testData/checkpoint"); // 添加数据源(这里使用的是自定义数据源CustomizeSource,方便测试) DataStreamSource<CustomizeBean> sourceStream = env.addSource(new CustomizeSource()); // 将数据流中的数据存储到bean对象中 SingleOutputStreamOperator<User> userMapStream = sourceStream.map(bean -> new User(bean.getName(), bean.getAge(), bean.getGender(), bean.getHobbit())); // 构建parquetWriterFactory ParquetWriterFactory<User> parquetWriterFactory2 = AvroParquetWriters.forReflectRecord(User.class); // 构建FileSink FileSink<User> parquetFileSink = FileSink // 使用Bulk模式,并配置路径和对应的schema .forBulkFormat(new Path("/Users/xxx/data/testData/"), parquetWriterFactory2) // 分桶策略,使用默认的 .withBucketAssigner(new DateTimeBucketAssigner<User>()) // 每100毫秒检查一次分桶 .withBucketCheckInterval(100) // 滚动策略,Bulk的滚动策略只有一种,就是发生Checkpoint的时候才进行滚动(为了保证列式文件的完整性) .withRollingPolicy(OnCheckpointRollingPolicy.build()) .build(); // 输出到文件 userMapStream.sinkTo(parquetFileSink); env.execute(); } } @Getter @Setter @ToString @NoArgsConstructor @AllArgsConstructor class User { private String name; private int age; private String gender; private String hobbit; }
代码中注释很详细了,具体使用看注释即可。这里说明一下为什么
forBulkFormat
的滚动策略只有OnCheckpointRollingPolicy
而不是像forRowFormat
那样可以通过时间和文件大小来控制文件滚动,注释中我也讲了是为了保证列式存储文件的完整性,因为列式文件中记录了很多信息,并不想行式存储文件一行一行的写就行,写到某一行直接停了也不影响文件的使用,而列式存储文件中不单单是记录了数据本身还有对应的字段类型、文件头信息、文件尾信息、切片索引等很多信息,如果在写入数据时某一刻直接停止了,而文件还没有生成完整的信息那就会导致这个烈士存储文件根本不具备使用性,是无法进行解析的。就比如说
ParquetFile
,它的文件结构如下图
可以看到文件的结构信息是很复杂的,如果感兴了解一下可以看数据存储格式这篇文章了解一下,这里就不细说了,内容还是比较多的.
-
AvroParquetWriters.forSpecificRecord(方式二)
forSpecificRecord
的使用不像forReflectRecord
那样自定义一个bean
接收数据就行了,使用forSpecificRecord
还要结合一下Apache avro
的官网看一下,下面我就介绍一下如何使用forSpecificRecord
.avro
的使用有两种方式一是通过API
直接调用的方式,二通过配置avsc
文件然后进行编译的方式,在代码中我们使用的第二种方式,使用第一种方式同样会出现很多schema
的信息在代码中写死修改起来会比较复杂的问题,而且对avro
的API
也要足够熟悉,学习成本还是有的.-
在
resource
目录中创建avsc
文件,文件内容如下{ "namespace": "com.jin.schema", "type": "record", "name": "UserSchemaBean", "fields": [ {"name": "name", "type": "string"}, {"name": "age", "type": "int"}, {"name": "gender", "type": "string"}, {"name": "hobbit", "type": "string"} ] }
文件中的内容就是
schema
信息,这里我相信大家都能看得明白."namespace": "com.jin.schema"
编译后自动创建的bean
的存储位置,"name": "UserSchemaBean"
就是配置生成bean
的名称,fields
中就是配置生成bean
的成员变量和对应的数据类型.官网演示的
avsc
文件内容如下:{"namespace": "example.avro", "type": "record", "name": "User", "fields": [ {"name": "name", "type": "string"}, {"name": "favorite_number", "type": ["int", "null"]}, {"name": "favorite_color", "type": ["string", "null"]} ] }
编译后就会根据
avsc
文件中的schema
信息在配置好的目录中自动创建bean
. -
在
Maven
中添加avsc
文件编译插件官网内容如下:
<plugin> <groupId>org.apache.avro</groupId> <artifactId>avro-maven-plugin</artifactId> <version>1.11.1</version> <executions> <execution> <phase>generate-sources</phase> <goals> <goal>schema</goal> </goals> <configuration> <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory> <outputDirectory>${project.basedir}/src/main/java/</outputDirectory> </configuration> </execution> </executions> </plugin>
要注意
<sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
是已经配置完的avsc
文件的位置,像是我就是在原有的resource
目录下配置的就要将内容改成<sourceDirectory>${project.basedir}/src/main/resource/</sourceDirectory>
否则在编译时就会报错找不到对应的目录或文件,如果想直接使用<sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
那就在项目的main
目录下创建一个avro
目录并将目录性质改为Source root
(这个如果不会可自行百度,关键字我都已经提供了).我的项目中实际配置如下:
<!-- avro插件 --> <plugin> <groupId>org.apache.avro</groupId> <artifactId>avro-maven-plugin</artifactId> <version>1.10.0</version> <executions> <execution> <phase>generate-sources</phase> <goals> <goal>schema</goal> </goals> <configuration> <sourceDirectory>${project.basedir}/src/main/resources/</sourceDirectory> <outputDirectory>${project.basedir}/src/main/java/</outputDirectory> </configuration> </execution> </executions> </plugin>
选择插件的版本时要注意依赖冲突问题,我们要先看一下Flink的
flink-avro
下的org.apache.avro:avro
是什么版本,如下图:
可以看到
1.15.3
的org.apache.avro:avro
的版本是1.10.0
,所以我选择的插件也是这个版本. -
编译
上面步骤都完成了就可以进行编译了,Maven->Lifecycle->compile,这里看一下编译后的结果如下图:
可以看到已经根据我们配置的
avsc
文件自动创建了对应的bean
,这里看一下成员变量内容是否一致,如下:/** * All-args constructor. * @param name The new value for name * @param age The new value for age * @param gender The new value for gender * @param hobbit The new value for hobbit */ public UserSchemaBean(java.lang.CharSequence name, java.lang.Integer age, java.lang.CharSequence gender, java.lang.CharSequence hobbit) { this.name = name; this.age = age; this.gender = gender; this.hobbit = hobbit; }
可以看到成员变量信息也是完全一致,我这里值展示了小部分代码,编译后的
bean
中的代码信息很多,不过我们不用关心这个,懂与不懂都不影响使用. -
代码内容
接下来就到主题了,实际的代码内容如下:
import com.jin.schema.UserSchemaBean; import org.apache.flink.connector.file.sink.FileSink; import org.apache.flink.core.fs.Path; import org.apache.flink.formats.parquet.ParquetWriterFactory; import org.apache.flink.formats.parquet.avro.AvroParquetWriters; import org.apache.flink.streaming.api.CheckpointingMode; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.CheckpointConfig; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner; import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy; /** * @Author: J * @Version: 1.0 * @CreateTime: 2023/6/28 * @Description: 测试 **/ public class FlinkFileSinkForParquet { public static void main(String[] args) throws Exception { // 创建流环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 设置并行度 env.setParallelism(1); // 每30秒作为checkpoint的一个周期 env.enableCheckpointing(30000); // 两次checkpoint间隔最少是20秒 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(20000); // 程序取消或者停止时不删除checkpoint env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // checkpoint必须在60秒结束,否则将丢弃 env.getCheckpointConfig().setCheckpointTimeout(60000); // 同一时间只能有一个checkpoint env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); // 设置EXACTLY_ONCE语义,默认就是这个 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); // checkpoint存储位置 env.getCheckpointConfig().setCheckpointStorage("file:///Users/xxx/data/testData/checkpoint"); // 添加数据源(这里使用的是自定义数据源,方便测试) DataStreamSource<CustomizeBean> sourceStream = env.addSource(new CustomizeSource()); // 将数据流中的对象转成UserSchemaBean类型 SingleOutputStreamOperator<UserSchemaBean> mapStream = sourceStream.map(bean -> new UserSchemaBean(bean.getName(), bean.getAge(), bean.getGender(), bean.getHobbit())); // 构建parquetWriterFactory,这里传入的就是编译后的UserSchemaBean ParquetWriterFactory<UserSchemaBean> parquetWriterFactory = AvroParquetWriters.forSpecificRecord(UserSchemaBean.class); // 构建FileSink FileSink<UserSchemaBean> parquetFileSink = FileSink // 使用Bulk模式,并配置路径和对应的schema .forBulkFormat(new Path("/Users/xxx/data/testData/"), parquetWriterFactory) // 分桶策略,使用默认的 .withBucketAssigner(new DateTimeBucketAssigner<UserSchemaBean>()) // 每100毫秒检查一次分桶 .withBucketCheckInterval(100) // 滚动策略,Bulk的滚动策略只有一种,就是发生Checkpoint的时候才进行滚动(为了保证列式文件的完整性) .withRollingPolicy(OnCheckpointRollingPolicy.build()) .build(); // 输出到文件 mapStream.sinkTo(parquetFileSink); env.execute(); } }
通过代码我们可以看到,内容基本就是一致的无非就是
forSpecificRecord
传入的bean
不同而已,当然还是建议使用AvroParquetWriters.forReflectRecord
这种方式,简易高效,复杂的过程并不一定能提高我们的代码能力.到这里这两种方式我都介绍完了,希望看完这篇文章有所收获.
-