azurehdinsight的sparkrunner上的apachebeam管道

dvtswwa3  于 2021-06-01  发布在  Hadoop
关注(0)|答案(1)|浏览(388)

我尝试在azure的hdinsight sparkrunner上运行一个beam管道。我首先尝试了一个基于spark2.3.0/hadoop2.7（hdi3.6）和2.3.1/hadoop3.0（hdi4.0预览版）的集群。我尝试使用ApacheBeam2.2.0和Next2.10.0-snapshot。
spark submit命令是（对于beam 2.10.0）：

JARS="wasbs:///dependency/hadoop-azure-3.1.1.3.0.2.0-50.jar,wasbs:///dependency/azure-storage-7.0.0.jar,wasbs:///dependency/beam-model-fn-execution-2.10.0-SNAPSHOT.jar,wasbs:///dependency/beam-model-job-management-2.10.0-SNAPSHOT.jar,wasbs:///dependency/beam-model-pipeline-2.10.0-SNAPSHOT.jar,wasbs:///dependency/beam-runners-core-construction-java-2.10.0-SNAPSHOT.jar,wasbs:///dependency/beam-runners-core-java-2.10.0-SNAPSHOT.jar,wasbs:///dependency/beam-runners-direct-java-2.10.0-SNAPSHOT.jar,wasbs:///dependency/beam-runners-spark-2.10.0-SNAPSHOT.jar,wasbs:///dependency/beam-sdks-java-core-2.10.0-SNAPSHOT.jar,wasbs:///dependency/beam-sdks-java-fn-execution-2.10.0-SNAPSHOT.jar,wasbs:///dependency/beam-sdks-java-io-hadoop-file-system-2.10.0-SNAPSHOT.jar,wasbs:///dependency/beam-vendor-grpc-1_13_1-0.1.jar"

spark-submit --conf spark.yarn.maxAppAttempts=1 --deploy-mode cluster --master yarn --jars $JARS --class example.MinimalWordCountJava8 wasbs:///mavenproject1-1.0-SNAPSHOT.jar --runner=SparkRunner

（最初，没有给jars提供hadoopazure和azure存储jar，但这没有任何区别）。
这个 main() 看起来像这样：

public static void main(String[] args) {

    JavaSparkContext ct = new JavaSparkContext();
    Configuration config = ct.hadoopConfiguration();

    config.set("fs.wasbs.impl", "org.apache.hadoop.fs.azure.NativeAzureFileSystem");
    config.set("fs.wasb.impl",  "org.apache.hadoop.fs.azure.NativeAzureFileSystem");
    config.set("fs.AbstractFileSystem.wasb.impl", "org.apache.hadoop.fs.azure.Wasb");
    config.set("fs.AbstractFileSystem.wasb.impl", "org.apache.hadoop.fs.azure.Wasbs");
    config.set("fs.azure", "org.apache.hadoop.fs.azure.NativeAzureFileSystem");
    config.set("fs.azure.account.key." + account + ".blob.core.windows.net", key);
    config.set("fs.defaultFS", "wasb://" + container + "@" + account + ".blob.core.windows.net");

    System.out.println("### hello.txt content:");
    JavaRDD<String> content = ct.textFile("wasbs:///hello.txt");
    System.out.println(content.toString());

    System.out.println("### MinimalWordCountJava8");

    PipelineOptions options = PipelineOptionsFactory.create();
    SparkContextOptions sparkContextOptions = options.as(SparkContextOptions.class);
    sparkContextOptions.setUsesProvidedSparkContext(true);
    sparkContextOptions.setProvidedSparkContext(ct);
    sparkContextOptions.setRunner(SparkRunner.class);

    Pipeline p = Pipeline.create(sparkContextOptions);

    p.apply(TextIO.read().from("hello.txt"))
     .apply(FlatMapElements
         .into(TypeDescriptors.strings())
         .via((String word) -> Arrays.asList(word.split("[^\\p{L}]+"))))
     .apply(Filter.by((String word) -> !word.isEmpty()))
     .apply(Count.<String>perElement())
     .apply(MapElements
         .into(TypeDescriptors.strings())
         .via((KV<String, Long> wordCount) -> wordCount.getKey() + ": " + wordCount.getValue()))
     // CHANGE 3/3: The Google Cloud Storage path is required for outputting the results to.
     .apply(TextIO.write().to("output"));

    p.run().waitUntilFinish();

调用时失败 Pipeline.create(options); 对于此异常跟踪：

18/12/09 14:47:10 ERROR ApplicationMaster: User class threw exception: java.lang.IllegalArgumentException: Failed to construct Hadoop filesystem with configuration Configuration: /usr/hdp/3.0.2.0-50/hadoop/conf/core-site.xml, /usr/hdp/3.0.2.0-50/hadoop/conf/hdfs-site.xml
java.lang.IllegalArgumentException: Failed to construct Hadoop filesystem with configuration Configuration: /usr/hdp/3.0.2.0-50/hadoop/conf/core-site.xml, /usr/hdp/3.0.2.0-50/hadoop/conf/hdfs-site.xml
    at org.apache.beam.sdk.io.hdfs.HadoopFileSystemRegistrar.fromOptions(HadoopFileSystemRegistrar.java:59)
    at org.apache.beam.sdk.io.FileSystems.verifySchemesAreUnique(FileSystems.java:489)
    at org.apache.beam.sdk.io.FileSystems.setDefaultPipelineOptions(FileSystems.java:479)
    at org.apache.beam.sdk.PipelineRunner.fromOptions(PipelineRunner.java:47)
    at org.apache.beam.sdk.Pipeline.create(Pipeline.java:145)
    at io.aptly.mavenproject1.MinimalWordCountJava8.main(MinimalWordCountJava8.java:88)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.apache.spark.deploy.yarn.ApplicationMaster$$anon$4.run(ApplicationMaster.scala:721)
Caused by: org.apache.hadoop.fs.UnsupportedFileSystemException: No FileSystem for scheme "wasbs"
    at org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:3332)
    at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3352)
    at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:124)
    at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3403)
    at org.apache.hadoop.fs.FileSystem$Cache.getUnique(FileSystem.java:3377)
    at org.apache.hadoop.fs.FileSystem.newInstance(FileSystem.java:530)
    at org.apache.hadoop.fs.FileSystem.newInstance(FileSystem.java:542)
    at org.apache.beam.sdk.io.hdfs.HadoopFileSystem.<init>(HadoopFileSystem.java:82)
    at org.apache.beam.sdk.io.hdfs.HadoopFileSystemRegistrar.fromOptions(HadoopFileSystemRegistrar.java:56)
    ... 10 more
18/12/09 14:47:10 INFO ApplicationMaster: Final app status: FAILED, exitCode: 15, (reason: User class threw exception: java.lang.IllegalArgumentException: Failed to construct Hadoop filesystem with configuration Configuration: /usr/hdp/3.0.2.0-50/hadoop/conf/core-site.xml, /usr/hdp/3.0.2.0-50/hadoop/conf/hdfs-site.xml
    at org.apache.beam.sdk.io.hdfs.HadoopFileSystemRegistrar.fromOptions(HadoopFileSystemRegistrar.java:59)
    at org.apache.beam.sdk.io.FileSystems.verifySchemesAreUnique(FileSystems.java:489)
    at org.apache.beam.sdk.io.FileSystems.setDefaultPipelineOptions(FileSystems.java:479)
    at org.apache.beam.sdk.PipelineRunner.fromOptions(PipelineRunner.java:47)
    at org.apache.beam.sdk.Pipeline.create(Pipeline.java:145)
    at io.aptly.mavenproject1.MinimalWordCountJava8.main(MinimalWordCountJava8.java:88)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.apache.spark.deploy.yarn.ApplicationMaster$$anon$4.run(ApplicationMaster.scala:721)
Caused by: org.apache.hadoop.fs.UnsupportedFileSystemException: No FileSystem for scheme "wasbs"
    at org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:3332)
    at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3352)
    at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:124)
    at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3403)
    at org.apache.hadoop.fs.FileSystem$Cache.getUnique(FileSystem.java:3377)
    at org.apache.hadoop.fs.FileSystem.newInstance(FileSystem.java:530)
    at org.apache.hadoop.fs.FileSystem.newInstance(FileSystem.java:542)
    at org.apache.beam.sdk.io.hdfs.HadoopFileSystem.<init>(HadoopFileSystem.java:82)
    at org.apache.beam.sdk.io.hdfs.HadoopFileSystemRegistrar.fromOptions(HadoopFileSystemRegistrar.java:56)

提交工程（the submit works） wasps:// 是公认的）和阅读的小 wasps:///hello.txt 不会失败。这些案例表明 wasps:// 在那之前一切都很好。
这是早期内梁，它似乎失败了。
因此我通过了考试 JavaSparkContext 与 PipelineOptions （使用其他so问题/答案建议的动态hadoop配置）。但这对我来说没什么不同。
有谁能指导如何绕过这个问题？

hadoopapache-sparkapache-beamazure-hdinsightazure-storage

1条答案

按热度按时间
kq0g1dla

kq0g1dla1#

从快速挖掘代码和bug跟踪器来看，从hadoop3.2.0（代码，jira）开始，azure似乎作为hadoop文件系统得到了支持。当前梁固定到版本2.7.3。这就解释了梁的失效 HadoopFilesystem .
可能是这样的 spark-submit 成功是因为 wasbs:// 通过与hadoop库不同的机制或使用捆绑的更新版本的hadoop来支持。

赞(0）分享  回复(0）举报  2021-06-01
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com