Spark 从 Blob 读取文本文件答案

【问题标题】：Spark reading text file from BlobSpark 从 Blob 读取文本文件
【发布时间】：2019-11-12 04:06:18
【问题描述】：

编写了一个代码以通过 spark 读取文本文件...在本地工作正常...但在 HDInsight 中运行时会产生错误 -> 从 Blob 读取文本文件

org.apache.spark.SparkException：作业因阶段失败而中止：阶段 0.0 中的任务 0 失败 4 次，最近一次失败：阶段 0.0 中丢失任务 0.3（TID 5，wn1-hchdin.bpqkkmavxs0ehkfnaruw4ed03d.dx.internal .cloudapp.net, 执行器 2): java.lang.AbstractMethodError: com.journaldev.sparkdemo.WordCounter$$Lambda$17/790636414.call(Ljava/lang/Object;)Ljava/util/Iterator; 在 org.apache.spark.api.java.JavaRDDLike$$anonfun$fn$1$1.apply(JavaRDDLike.scala:125) 在 org.apache.spark.api.java.JavaRDDLike$$anonfun$fn$1$1.apply(JavaRDDLike.scala:125) 在 scala.collection.Iterator$$anon$12.nextCur(Iterator.scala:434) 在 scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440) 在 scala.collection.Iterator$class.foreach(Iterator.scala:893) 在 scala.collection.AbstractIterator.foreach(Iterator.scala:1336) 在 org.apache.spark.rdd.RDD$$anonfun$foreach$1$$anonfun$apply$28.apply(RDD.scala:927) 在 org.apache.spark.rdd.RDD$$anonfun$foreach$1$$anonfun$apply$28.apply(RDD.scala:927) 在 org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074) 在 org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074) 在 org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87) 在 org.apache.spark.scheduler.Task.run(Task.scala:109) 在 org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) 在 java.lang.Thread.run(Thread.java:748)

这是我的代码

    JavaSparkContext ct = new JavaSparkContext();
        Configuration config = ct.hadoopConfiguration();
        config.set("fs.azure", "org.apache.hadoop.fs.azure.NativeAzureFileSystem");
        config.set("org.apache.hadoop.fs.azure.SimpleKeyProvider", "<<key>>");

        JavaRDD<String> inputFile = ct.textFile("wasb://<<container-nam>>@<<account>>.blob.core.windows.net/directory/file.txt");

        JavaRDD<String> wordsFromFile = inputFile.flatMap(content -> Arrays.asList(content.split(" ")));

        wordsFromFile.foreach(cc ->{System.out.println("p :"+cc.toString());});

【问题讨论】：

请看下面的帖子是否有帮助stackoverflow.com/questions/37763472/…

标签： java apache-spark blob azure-hdinsight

【解决方案1】：

对于在本地运行的 Spark，有一个官方的blog 介绍了如何从 Spark 访问 Azure Blob Storage。关键是您需要在 core-site.xml 文件中将 Azure 存储帐户配置为与 HDFS 兼容的存储，并将两个 jars hadoop-azure 和 azure-storage 添加到您的类路径中，以便通过协议 wasb[s] 访问 HDFS。您可以参考官方tutorial了解使用wasb的HDFS兼容存储，以及blog关于HDInsight配置的更多详细信息。

对于运行在 Azure 上的 Spark，区别只是使用 wasb 访问 HDFS，其他准备工作在使用 Spark 创建 HDInsight 集群时由 Azure 完成。列出文件的方法是SparkContext的listFiles或wholeTextFiles。

希望对你有帮助。

【讨论】：