HDFS Config for Pyspark

Question

I' tryng to read a file from HDFS using pyspark. The code is the following:

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()

using the command:

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

But I got the following Error:

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)

I believe that there is a problem in the configuration of HDFS for pyspark. (I don't have hdfs-site.xml file in my Hadoop folder) What am I missing?

Thank you

EDIT: I resolved! The problem was inside the etc/hosts file: when you use pyspark you must add ALL the IP of namenode and datanode. Now it works.

Answer 1

Change your code as follows.

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()

When reading files from a location, one must simply provide the path till the parent folder that contains the data. Beyound this, the Spark DataframeReader class can load the avro files since you have used the avro method while reading the the HDFS path