前言
ZSTD压缩算法是现在最流行压缩算法了，有着高压缩比和压缩性能强的有点，已经被各大框架所使用。
目前hadoop 3.1.0版本已经支持ZSTD算法。所以可以使用Flink写HDFS时使用这个算法。具体如何操作请往下看
升级hadoop版本
Flink1.11已经支持hadoop 3.x+版本了，所以Flink依赖的hadoop shaded包版本不达标的需要升级。
由于从Flink1.11开始，官方已经不提供hadoop shaded支持，可以使用CLASS_PATH方式 参考。
但是任然可以自己编译hadoop 3.1版本的shaded包，具体步骤如下：
- git clone https://github.com/apache/flink-shaded.git
- 切换到release-10.0，由于flink1.11 已经不再支持hadoop shaded包，所以需要我们自己打包
- 修改flink-shaded-hadoop-2-parent的pom.xml版本为11.0、hadoop.version 为3.1.0
- 修改flink-shaded-hadoop-2的pom.xml parent version为11.0、version为${hadoop.version}-11.0
- 修改flink-shaded-hadoop-2-uber的pom.xml parent version为11.0、version为
{hadoop.version}-11.0
- 在根目录下执行命令
mvn clean install -DskipTests -Dcheckstyle.skip=true -Dhadoop.version=3.1.0 - 完成后将flink-shaded-hadoop-2-uber-3.1.0-11.0.jar放到flink lib目录下
hadoop集群支持
同时hadoop集群版本也要支持ZSTD，升级集群或者将ZSTD功能merge到低版本中。
SQL
现在就可以写flink任务了，DDL如下：
CREATE TABLE flink_filesystem_parquet_zstd (
a BIGINT,
b STRING,
c STRING,
d STRING,
e STRING
) PARTITIONED BY (e) WITH (
'connector'='filesystem',
'path' = 'hdfs://xxx',
'format'='parquet',
'hadoop-user'='hadoop',
'parquet.compression'='ZSTD'
);
遇到的问题
- NoClassDefException org.apache.hadoop.io.compress.ZStandardCodec
原因：依赖的flink shaded hadoop包是2.7.2版本，这个类是hadoop-common 3.1.0中的类，所以需要升级hadoop包到3.1.0 - shaded 包升级之后执行报错
2020-10-21 15:08:14
java.lang.UnsatisfiedLinkError: org.apache.hadoop.util.NativeCodeLoader.buildSupportsZstd()Z
at org.apache.hadoop.util.NativeCodeLoader.buildSupportsZstd(Native Method)
at org.apache.hadoop.io.compress.ZStandardCodec.checkNativeCodeLoaded(ZStandardCodec.java:64)
at org.apache.hadoop.io.compress.ZStandardCodec.getCompressorType(ZStandardCodec.java:153)
at org.apache.hadoop.io.compress.CodecPool.getCompressor(CodecPool.java:150)
at org.apache.hadoop.io.compress.CodecPool.getCompressor(CodecPool.java:168)
at org.apache.parquet.hadoop.CodecFactory$HeapBytesCompressor.<init>(CodecFactory.java:144)
at org.apache.parquet.hadoop.CodecFactory.createCompressor(CodecFactory.java:206)
at org.apache.parquet.hadoop.CodecFactory.getCompressor(CodecFactory.java:189)
at org.apache.parquet.hadoop.ParquetWriter.<init>(ParquetWriter.java:285)
at org.apache.parquet.hadoop.ParquetWriter$Builder.build(ParquetWriter.java:530)
at org.apache.flink.formats.parquet.row.ParquetPb2AvroBuilder$FlinkParquetPb2AvroBuilder.createWriter(ParquetPb2AvroBuilder.java:212)
at org.apache.flink.formats.parquet.ParquetWriterFactory.create(ParquetWriterFactory.java:57)
at org.apache.flink.table.filesystem.FileSystemTableSink$ProjectionBulkFactory.create(FileSystemTableSink.java:521)
at org.apache.flink.streaming.api.functions.sink.filesystem.BulkBucketWriter.openNew(BulkBucketWriter.java:69)
at org.apache.flink.streaming.api.functions.sink.filesystem.OutputStreamBasedPartFileWriter$OutputStreamBasedBucketWriter.openNewInProgressFile(OutputStreamBasedPartFileWriter.java:83)
at org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.rollPartFile(Bucket.java:209)
at org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.write(Bucket.java:200)
at org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.onElement(Buckets.java:282)
at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.onElement(StreamingFileSinkHelper.java:104)
at org.apache.flink.table.filesystem.stream.StreamingFileWriter.processElement(StreamingFileWriter.java:127)
at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:161)
at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.processElement(StreamTaskNetworkInput.java:178)
at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.emitNext(StreamTaskNetworkInput.java:153)
at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:67)
at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:345)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxStep(MailboxProcessor.java:191)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:181)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:558)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:530)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546)
at java.lang.Thread.run(Thread.java:748)
原因：线上环境是hadoop2.6版本 还不支持zstd压缩，之后在一个测试环境上进行，该环境还是2.6版本但是将3.1的特性merge了进来。
