Reading multiple text files using Spark

Question

I am working in Spark trying to read in multiple text files from a single directory. I have read multiple tutorials and Q&A sessions on this topic and it should be straightforward. In spite of that, I still cannot get it to work on my system.

I am working on Windows 10 with Python 8.5, Java 8, and Anaconda 3.

In order to keep my testing really simple, all I am trying to do is load the text files into an RDD and print out the contents. Here are my experiments with what worked and what did not work:

#THESE WORK

#files = sc.textFile("C:/spark/HW1/data/199901.txt,C:/spark/HW1/data/200002.txt,C:/spark/HW1/data/200404.txt,C:/spark/HW1/data/201003.txt")

#files = sc.textFile("data/199901.txt,data/200002.txt,data/200404.txt,data/201003.txt")

#files = sc.wholeTextFiles("data/199901.txt,data/200002.txt,data/200404.txt,data/201003.txt")

#THESE FAIL

#files = sc.textFile("*.txt")

#files = sc.textFile("data/*.txt")

#files = sc.wholeTextFiles("data")

#files = sc.wholeTextFiles("C:/spark/HW1/data/*")

#files = sc.textFile("C:/spark/HW1/data")

In short, explicit file lists work fine, but wildcards or just naming the directory does not. While explicit lists are fine for small cases, I need this to work on many files, so explicit will not work in the long run.

For all the failed cases, here are the follow on steps and the errors:

llist = files.collect()

for line in llist:
    print(line)

Traceback (most recent call last):
  File "C:/spark/HW1/test2.py", line 30, in 
    llist = files.collect()
  File "c:\spark\python\lib\pyspark.zip\pyspark\rdd.py", line 1197, in collect
  File "c:\spark\python\lib\py4j-0.10.9.5-src.zip\py4j\java_gateway.py", line 1321, in __call__
  File "c:\spark\python\lib\py4j-0.10.9.5-src.zip\py4j\protocol.py", line 326, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
: java.lang.UnsatisfiedLinkError: org.apache.hadoop.io.nativeio.NativeIO$Windows.access0(Ljava/lang/String;I)Z
        at org.apache.hadoop.io.nativeio.NativeIO$Windows.access0(Native Method)
        at org.apache.hadoop.io.nativeio.NativeIO$Windows.access(NativeIO.java:793)
        at org.apache.hadoop.fs.FileUtil.canRead(FileUtil.java:1218)
        at org.apache.hadoop.fs.FileUtil.list(FileUtil.java:1423)
        at org.apache.hadoop.fs.RawLocalFileSystem.listStatus(RawLocalFileSystem.java:601)
        at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:1972)
        at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:2014)
        at org.apache.hadoop.fs.FileSystem$4.(FileSystem.java:2180)
        at org.apache.hadoop.fs.FileSystem.listLocatedStatus(FileSystem.java:2179)
        at org.apache.hadoop.fs.ChecksumFileSystem.listLocatedStatus(ChecksumFileSystem.java:783)
        at org.apache.hadoop.mapred.FileInputFormat.singleThreadedListStatus(FileInputFormat.java:285)
        at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:244)
        at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:332)
        at org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:208)
        at org.apache.spark.rdd.RDD.$anonfun$partitions$2(RDD.scala:292)
        at scala.Option.getOrElse(Option.scala:189)
        at org.apache.spark.rdd.RDD.partitions(RDD.scala:288)
        at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:49)
        at org.apache.spark.rdd.RDD.$anonfun$partitions$2(RDD.scala:292)
        at scala.Option.getOrElse(Option.scala:189)
        at org.apache.spark.rdd.RDD.partitions(RDD.scala:288)
        at org.apache.spark.SparkContext.runJob(SparkContext.scala:2293)
        at org.apache.spark.rdd.RDD.$anonfun$collect$1(RDD.scala:1021)
        at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
        at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
        at org.apache.spark.rdd.RDD.withScope(RDD.scala:406)
        at org.apache.spark.rdd.RDD.collect(RDD.scala:1020)
        at org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:180)
        at org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)
        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.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)
        at py4j.ClientServerConnection.run(ClientServerConnection.java:106)
        at java.lang.Thread.run(Thread.java:750)

It is probably a syntax thing, but after all these test cases, I have no idea what it should be. Any help would be appreciated.

CRAFTY DBA · Answer

Using the spark context is just one way to read files. I usually use the pyspark library to read and write files. Just make sure no unwanted files are in the directory.

Here is a sample exercise. I have S&P 500 stock data for each day during 2013 to 2017. Normally, I would read this as csv files but will read as text for your example.

The image shows some files for the year 2017.

This storage is mounted to my Azure Databricks cluster. The image below shows two shell commands. The first one shows there are 2533 files in all sub-directories. The second shows there are 5 directories.

The code below reads in all files in the given path as text. It uses a recursive option to grab all files.

# import function
from  pyspark.sql.functions import input_file_name

# path to files
path = "/mnt/datalake/bronze/stocks/"

# read in data
df1 = spark.read.option("recursiveFileLookup", "true").text(path)

# add file name as column
df1 = df1.withColumn("filename", input_file_name())

We can see that we are treating all data as lines. This includes the header. The total number of files is just over 640K lines.

Let us write some simple aggregations to make sure we have the correct number of files.

This simple aggregation by file name shows the number of lines in each file. It takes the original dataframe (df1) and saves it as new dataframe (df2).

Last but not least, if we count the number of rows in df2, we get the same number of files that were initially on my pc, then uploaded to ADLS Gen2, and the read in by Azure Databricks.

This same code spark.read() should work on any spark system.

Happy programming.

Reading multiple text files using Spark

1 Answers