Flink StreamingFileSink<GenericRecord> 和 Azure Datalake Gen 2 的问题

Question

我在尝试使用 Flink 的 StreamingFileSink 将文件放入 Azure Datalake Gen 2 时遇到问题，我正在使用 core-site.xml 和 Hadoop 批量格式我正在尝试使用 abfss:// 格式复制到我的数据湖（也尝试使用 abfs://）

java.lang.UnsupportedOperationException: Recoverable writers on Hadoop are only supported for HDFS
[job-playground-job-cluster-0 flink-job-cluster]        at org.apache.flink.runtime.fs.hdfs.HadoopRecoverableWriter.<init>(HadoopRecoverableWriter.java:61) ~[flink-dist_2.11-1.11.0.jar:1.11.0]
[job-playground-job-cluster-0 flink-job-cluster]        at org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.createRecoverableWriter(HadoopFileSystem.java:202) ~[flink-dist_2.11-1.11.0.jar:1.11.0]
[job-playground-job-cluster-0 flink-job-cluster]        at org.apache.flink.core.fs.SafetyNetWrapperFileSystem.createRecoverableWriter(SafetyNetWrapperFileSystem.java:69) ~[flink-dist_2.11-1.11.0.jar:1.11.0]
[job-playground-job-cluster-0 flink-job-cluster]        at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink$BulkFormatBuilder.createBuckets(StreamingFileSink.java:371) ~[flink-dist_2.11-1.11.0.jar:1.11.0]

我阅读了官方文档并深入研究了库，问题就在这里： https://github.com/apache/flink/blob/master/flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableWriter.java#L60

public HadoopRecoverableWriter(org.apache.hadoop.fs.FileSystem fs) {
        this.fs = checkNotNull(fs);

        // This writer is only supported on a subset of file systems
        if (!"hdfs".equalsIgnoreCase(fs.getScheme())) {
            throw new UnsupportedOperationException(
                    "Recoverable writers on Hadoop are only supported for HDFS");
        }

这是我的核心-site.xml

<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>

<configuration>
  <property>
    <name>fs.azure.account.auth.type.ADLS_ACCOUNT_NAME.dfs.core.windows.net</name>
    <value>SharedKey</value>
    <description>
    It is inferred by the url
    </description>
  </property>

  <property>
    <name>fs.azure.account.key.ADLS_ACCOUNT_NAME.dfs.core.windows.net</name>
    <value>ADLS_KEY</value>
    <description>
    </description>
  </property>

  <property>
    <name>fs.azure.createRemoteFileSystemDuringInitialization</name>
    <value>true</value>
  </property>

  <property>
    <name>fs.azure.always.use.https</name>
    <value>true</value>
  </property>
  
</configuration>

任何人都通过了这个问题或者扩展有问题 abfss/abfs。

Answer 1

StreamingFileSink 不yet支持 Azure Data Lake。

Flink StreamingFileSink<GenericRecord> 和 Azure Datalake Gen 2 的问题

Problem with Flink StreamingFileSink<GenericRecord> & Azure Datalake Gen 2

java

apache-flink

flink-streaming

azure-data-lake-gen2