从 pyspark 读取 hdfs 中的文件答案

【问题标题】：reading a file in hdfs from pyspark从 pyspark 读取 hdfs 中的文件
【发布时间】：2016-06-20 12:36:58
【问题描述】：

我正在尝试读取我的 hdfs 中的文件。这是我的hadoop文件结构的展示。

hduser@GVM:/usr/local/spark/bin$ hadoop fs -ls -R /
drwxr-xr-x   - hduser supergroup          0 2016-03-06 17:28 /inputFiles
drwxr-xr-x   - hduser supergroup          0 2016-03-06 17:31 /inputFiles/CountOfMonteCristo
-rw-r--r--   1 hduser supergroup    2685300 2016-03-06 17:31 /inputFiles/CountOfMonteCristo/BookText.txt

这是我的 pyspark 代码：

from pyspark import SparkContext, SparkConf

conf = SparkConf().setAppName("myFirstApp").setMaster("local")
sc = SparkContext(conf=conf)

textFile = sc.textFile("hdfs://inputFiles/CountOfMonteCristo/BookText.txt")
textFile.first()

我得到的错误是：

Py4JJavaError: An error occurred while calling o64.partitions.
: java.lang.IllegalArgumentException: java.net.UnknownHostException: inputFiles

这是因为我的 sparkContext 设置不正确吗？我通过虚拟盒在 ubuntu 14.04 虚拟机中运行它。

我不确定我在这里做错了什么......

【问题讨论】：

标签： apache-spark hdfs pyspark

【解决方案1】：

由于您不提供权限 URI 应该如下所示：

hdfs:///inputFiles/CountOfMonteCristo/BookText.txt

否则inputFiles 被解释为主机名。使用正确的配置，您根本不需要使用方案：

/inputFiles/CountOfMonteCristo/BookText.txt

改为。

【讨论】：

【解决方案2】：

如果没有提供配置，您可以通过完整路径访问 HDFS 文件。（如果 hdfs 位于本地环境中，namenodehost 是您的本地主机）。

hdfs://namenodehost/inputFiles/CountOfMonteCristo/BookText.txt

【讨论】：

有没有办法设置名称节点主机，这样它就不会在 python 文件中硬编码？我们怎样才能最好地参数化它？也许使用可以在许多应用程序之间共享的某种配置文件？

【解决方案3】：

在 Spark 中读取文件的一般方法有两种，一种用于并行处理大型分布式文件，另一种用于读取 HDFS 上的查找表和配置等小文件。对于后者，您可能希望将驱动程序节点或工作程序中的文件作为单次读取（而不是分布式读取）读取。在这种情况下，您应该使用SparkFiles 模块，如下所示。

# spark is a SparkSession instance
from pyspark import SparkFiles

spark.sparkContext.addFile('hdfs:///user/bekce/myfile.json')
with open(SparkFiles.get('myfile.json'), 'rb') as handle:
    j = json.load(handle)
    or_do_whatever_with(handle)

【讨论】：

【解决方案4】：

首先，你需要运行

export PYSPARK_PYTHON=python3.4 #what so ever is your python version

代码

from pyspark.sql import SparkSession
from pyspark import SparkConf, SparkContext

spark = SparkSession.builder.appName("HDFS").getOrCreate()
sparkcont = SparkContext.getOrCreate(SparkConf().setAppName("HDFS"))
logs = sparkcont.setLogLevel("ERROR")

data = [('First', 1), ('Second', 2), ('Third', 3), ('Fourth', 4), ('Fifth', 5)]
df = spark.createDataFrame(data)

df.write.csv("hdfs:///mnt/data/")
print("Data Written")

执行代码

spark-submit --master yarn --deploy-mode client <py file>

【讨论】：