pyspark的HDFS配置

Question

我尝试使用pyspark读取来自HDFS的文件。 代码如下：

import numpy as np
import pandas as pd 
from pyspark.sql import SparkSession
import json
import sys
import io   
import os

os.environ["HADOOP_USER_NAME"] = "hdfs"

spark = SparkSession.builder.master("local") \
                .appName('PySpark_Neural_Network') \
                .config("spark.hadoop.dfs.client.use.datanode.hostname", "true") \
                .config("spark.driver.memory", "16g")\
                .getOrCreate()


df = spark.read.format("avro").load("hdfs://localhost:8020/data/file.avro", header=True)
df.show()

使用命令：

spark-submit --packages org.apache.spark:spark-avro_2.12:3.1.2 script.py

但是我收到以下错误：

py4j.protocol.Py4JJavaError: An error occurred while calling o39.load.
: java.nio.channels.UnresolvedAddressException
    at sun.nio.ch.Net.checkAddress(Net.java:100)
    at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:620)
    at org.apache.hadoop.net.SocketIOWithTimeout.connect(SocketIOWithTimeout.java:192)
    at org.apache.hadoop.net.NetUtils.connect(NetUtils.java:531)
    at org.apache.hadoop.hdfs.DFSClient.newConnectedPeer(DFSClient.java:2939)
    at org.apache.hadoop.hdfs.client.impl.BlockReaderFactory.nextTcpPeer(BlockReaderFactory.java:821)
    at org.apache.hadoop.hdfs.client.impl.BlockReaderFactory.getRemoteBlockReaderFromTcp(BlockReaderFactory.java:746)
    at org.apache.hadoop.hdfs.client.impl.BlockReaderFactory.build(BlockReaderFactory.java:379)
    at org.apache.hadoop.hdfs.DFSInputStream.getBlockReader(DFSInputStream.java:644)
    at org.apache.hadoop.hdfs.DFSInputStream.blockSeekTo(DFSInputStream.java:575)
    at org.apache.hadoop.hdfs.DFSInputStream.readWithStrategy(DFSInputStream.java:757)
    at org.apache.hadoop.hdfs.DFSInputStream.read(DFSInputStream.java:829)
    at java.io.DataInputStream.read(DataInputStream.java:149)
    at org.apache.avro.mapred.FsInput.read(FsInput.java:54)
    at org.apache.avro.file.DataFileReader.openReader(DataFileReader.java:55)
    at org.apache.spark.sql.avro.AvroUtils$.$anonfun$inferAvroSchemaFromFiles$3(AvroUtils.scala:139)
    at org.apache.spark.util.Utils$.tryWithResource(Utils.scala:2622)
    at org.apache.spark.sql.avro.AvroUtils$.$anonfun$inferAvroSchemaFromFiles$1(AvroUtils.scala:137)
    at scala.collection.Iterator$$anon$10.next(Iterator.scala:459)
    at scala.collection.TraversableOnce.collectFirst(TraversableOnce.scala:148)
    at scala.collection.TraversableOnce.collectFirst$(TraversableOnce.scala:135)
    at scala.collection.AbstractIterator.collectFirst(Iterator.scala:1429)
    at org.apache.spark.sql.avro.AvroUtils$.inferAvroSchemaFromFiles(AvroUtils.scala:151)
    at org.apache.spark.sql.avro.AvroUtils$.$anonfun$inferSchema$3(AvroUtils.scala:60)
    at scala.Option.getOrElse(Option.scala:189)
    at org.apache.spark.sql.avro.AvroUtils$.inferSchema(AvroUtils.scala:59)
    at org.apache.spark.sql.avro.AvroFileFormat.inferSchema(AvroFileFormat.scala:58)
    at org.apache.spark.sql.execution.datasources.DataSource.$anonfun$getOrInferFileFormatSchema$11(DataSource.scala:209)
    at scala.Option.orElse(Option.scala:447)
    at org.apache.spark.sql.execution.datasources.DataSource.getOrInferFileFormatSchema(DataSource.scala:206)
    at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:419)
    at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:325)
    at org.apache.spark.sql.DataFrameReader.$anonfun$load$3(DataFrameReader.scala:307)
    at scala.Option.getOrElse(Option.scala:189)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:307)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:239)
    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 py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:282)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:748)

我相信HDFS对Pyspark的配置存在问题。 （我的Hadoop文件夹中没有HDFS-site.xml文件） 我想念什么？

谢谢

编辑：我解决了！问题在etc/hosts文件中：使用Pyspark时，必须添加Namenode和DataNode的所有IP。现在起作用。

Answer 1

更改您的代码如下。

from pyspark.sql import SparkSession  
import os

os.environ["HADOOP_USER_NAME"] = "hdfs"

spark = SparkSession.builder.master("local") \
                .appName('PySpark_Neural_Network') \
                .config("spark.hadoop.dfs.client.use.datanode.hostname", "true") \
                .config("spark.driver.memory", "16g")\
                .getOrCreate()


df = spark.read.format("avro").load("hdfs://localhost:8020/data/", header=True)
df.show()

从位置读取文件时，必须简单地提供路径，直到包含数据的父文件夹为止。 Beyound此，Spark DataFramEreader类可以加载AVRO文件，因为您在阅读HDFS路径时使用了avro方法