flink1.10.0版本flink-dist

1、开发中引入了如下依赖：

  org.apache.flink
  flink-streaming-java_2.11
  1.10.0
  provided

代码中使用如下api：

this.parquetFileSink = StreamingFileSink.forBulkFormat(new Path(this.readDataPath),
        ParquetAvroWriters.forBucketWriteRecord(schema, compression, blockSize))
        .withBucketAssigner(new BucketAssigner() {
            @Override
            public String getBucketId(BucketWriteRecord o, Context context) {
                if (o != null) {
                    return o.getBucketId();
                }
                return "nopartition";
            }

            @Override
            public SimpleVersionedSerializer getSerializer() {
                return SimpleVersionedStringSerializer.INSTANCE;
            }
        })
        .withOutputFileConfig(config)
        .build();

其中StreamingFileSink.forBulkFormat方法使用的是flink-streaming-jar_2.11-1.10.0.jar中的api。

然后打包，提交到flink集群。

2、flink集群环境：

安装flink时，会在在$FLlink_HOME/lib下存放了一些flink依赖，其中flink-dist_2.11-1.10.0.jar是一个shade包，包含了flink自身需要的依赖包（所以，我们在代码的pom.xml中对flink-core、flink-clients_2.11、flink-java、flink-streaming-java_2.11进行provided），其中就含有flink-streaming-jar_2.11-1.10.0.jar。

集群上提交flink任务后，flink会将业务jar包和$FLlink_HOME/lib下的jar包一起提交到yarn，然后运行。

将上面的业务jar包运行，flink任务报如下错误：

2022-03-02 15:24:57
java.lang.NoSuchMethodError: org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.forBulkFormat(Lorg/apache/flink/core/fs/Path;Lorg/apache/flink/api/common/serialization/BulkWriter$Factory;)Lorg/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink$BulkFormatBuilder;
	at com.abc.parquetsink.sink.ParquetBulkBladeSink.createStreamFileSink(ParquetBulkBladeSink.java:183)
	at com.abc.parquetsink.sink.ParquetBulkBladeSink.setRuntimeContext(ParquetBulkBladeSink.java:206)
	at org.apache.flink.api.common.functions.util.FunctionUtils.setFunctionRuntimeContext(FunctionUtils.java:50)
	at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.setup(AbstractUdfStreamOperator.java:83)
	at org.apache.flink.streaming.api.operators.SimpleOperatorFactory.createStreamOperator(SimpleOperatorFactory.java:75)
	at org.apache.flink.streaming.api.operators.StreamOperatorFactoryUtil.createOperator(StreamOperatorFactoryUtil.java:48)
	at org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:429)
	at org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:353)
	at org.apache.flink.streaming.runtime.tasks.OperatorChain.(OperatorChain.java:144)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:432)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:460)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:707)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:532)
	at java.lang.Thread.run(Thread.java:748)

3、问题分析：

根据报错信息，StreamingFileSink.forBulkFormat方法找不到，我们在flink集群的lib下，反编译flink-dist_2.11-1.10.0.jar包，如下：

在项目代码中，查看源代码：

其源码所在的jar：

如此可见：StreamingFileSink.forBulkFormat方法在flink-dist_2.11-1.10.0.jar包和flink-streaming-jar_2.11-1.10.0.jar中的不一致，所以会报上面的错。经过调研，我们在flink-dist_2.12-1.10.0.jar中找到了和flink-streaming-jar_2.11-1.10.0.jar一致的方法：

 如此，可以推断出是flink1.10发型版本的lib中flink-dist_2.11-1.10.0.jar打错了。其中scala版本弄错了。