A Guide to Read CSV and JSON using Apache Spark
Apache Spark is a multi-language engine for executing data engineering, data science, and machine learning on single-node machines or clusters. Spark SQL is an advanced distributed SQL work on structured tables or unstructured data such as JSON or images.
In this article, we will explore different input / ouput operations possible in the spark. We will read data from CSV and JSON files. To learn basics of Java refer to article Getting started with Spark.
We are going to start the apache spark using Java with local standalone feature. Lets have java-8 or 11 to avoid spark compatability issues. To have spark in the project, we need to have spark-core and spark-sql. Spark-core is to create the spark session and spark sql to have data frame operations. Given the complete pom below:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.java21</groupId>
<artifactId>java21-deltalake-app</artifactId>
<version>1.0.0-SNAPSHOT</version>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<java.version>8</java.version>
<maven.compiler.source>${java.version}</maven.compiler.source>
<maven.compiler.target>${java.version}</maven.compiler.target>
<maven.compiler.release>${java.version}</maven.compiler.release>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.4.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>3.4.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-core -->
<!--dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
<version>2.18.2</version>
</dependency-->
<dependency>
<groupId>io.delta</groupId>
<artifactId>delta-core_2.13</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-engine</artifactId>
<version>5.10.2</version>
<scope>test</scope>
</dependency>
<!-- https://mvnrepository.com/artifact/org.scala-lang/scala-library -->
<!--dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.13.15</version>
</dependency-->
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>3.2.5</version>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.4.2</version>
<configuration>
<archive>
<manifest>
<mainClass>com.java21.StandaloneSparkTester</mainClass>
</manifest>
</archive>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
Now we can start the spark session from the below code snippet having master as 'local[*]'. config is optional, to give memory requests like executor and driver memory requests. After that check the spark context is stopped or running to confirm spark is started successfully.
SparkSession spark = SparkSession.builder()
.appName("Spark Connectivity Test")
.master("local[*]")
.config("spark.executor.memory", "2g")
.config("spark.driver.memory", "2g")
.getOrCreate();
System.out.println("spark connectivity...");
System.out.println(spark.catalog());
// spark.conf().set("spark.scheduler.listenerbus.eventqueue.size", "200000");
System.out.println(spark.sparkContext().isStopped());
Now lets create a data frame by having struct type schema for each column. We can use RowFactory to create each row of the dataset. DataSet of rows will become the spark data frame by using spark create data frame.
StructType schema = new StructType(new StructField[] {
new StructField("id", DataTypes.IntegerType, false, Metadata.empty()),
new StructField("product", DataTypes.StringType, false, Metadata.empty()),
new StructField("price", DataTypes.DoubleType, false, Metadata.empty()),
new StructField("category", DataTypes.StringType, false, Metadata.empty())
});
// Create data
Row row1 = RowFactory.create(1, "14inch Laptop", 30.50d, "Compute Devices");
Row row2 = RowFactory.create(2, "15inch Desktop", 45.50d, "Compute Devices");
Row row3 = RowFactory.create(3, "Ink Jet Printer", 90.50d, "Accessories");
List<Row> data = Arrays.asList(row1, row2, row3);
// Create DataFrame
Dataset<Row> df = spark.createDataFrame(data, schema);
df.show();
Output:
24/12/08 20:58:00 INFO CodeGenerator: Code generated in 35.81301 ms
+-------+---+---------------+-----------------+---------------+
|summary| id| product| price| category|
+-------+---+---------------+-----------------+---------------+
| count| 3| 3| 3| 3|
| mean|2.0| null| 55.5| null|
| stddev|1.0| null|31.22498999199199| null|
| min| 1| 14inch Laptop| 30.5| Accessories|
| 25%| 1| null| 30.5| null|
| 50%| 2| null| 45.5| null|
| 75%| 3| null| 90.5| null|
| max| 3|Ink Jet Printer| 90.5|Compute Devices|
+-------+---+---------------+-----------------+---------------+
24/12/08 20:58:00 INFO CodeGenerator: Code generated in 15.595309 ms
After creating the dataframe, we have done the summary showcase.
df.summary().show();
24/12/08 20:58:00 INFO CodeGenerator: Code generated in 35.81301 ms
+-------+---+---------------+-----------------+---------------+
|summary| id| product| price| category|
+-------+---+---------------+-----------------+---------------+
| count| 3| 3| 3| 3|
| mean|2.0| null| 55.5| null|
| stddev|1.0| null|31.22498999199199| null|
| min| 1| 14inch Laptop| 30.5| Accessories|
| 25%| 1| null| 30.5| null|
| 50%| 2| null| 45.5| null|
| 75%| 3| null| 90.5| null|
| max| 3|Ink Jet Printer| 90.5|Compute Devices|
+-------+---+---------------+-----------------+---------------+
Then we did aggregation of the category column to get the total price of those categories.
Dataset<Row> aggdf = df.groupBy("category").sum("price");
aggdf.show();
24/12/08 20:58:00 INFO CodeGenerator: Code generated in 36.705315 ms
+---+---------------+-----+---------------+
| id| product|price| category|
+---+---------------+-----+---------------+
| 1| 14inch Laptop| 30.5|Compute Devices|
| 2| 15inch Desktop| 45.5|Compute Devices|
| 3|Ink Jet Printer| 90.5| Accessories|
+---+---------------+-----+---------------+
CSV Reader
Spark session will help us to read CSV file with the header and convert directly into the spark data frame with dataset of rows.
Dataset<Row> csvdf = spark.read()
.option("header", "true")
.option("encoding", "UTF-8")
.csv("/home/nagappan/Downloads/out.csv");
csvdf.show();
While reading the csv file, if header columns doesnt tally, it shows the warning and still it makes the data frame for us. Below is the sample output, after reading csv file.
24/12/08 20:58:04 WARN CSVHeaderChecker: CSV header does not conform to the schema.
Header: RetailStoreID, OrganizationHierarchy, WorkstationID, InvoiceNO, BusinessDayDate, BeginDateTime, EndDateTime, OperatorID, OperatorName, CurrencyCode, TransactionTypeID,
Schema: RetailStoreID, OrganizationHierarchy, WorkstationID, InvoiceNO, BusinessDayDate, BeginDateTime, EndDateTime, OperatorID, OperatorName, CurrencyCode, TransactionTypeID, _c11
Expected: _c11 but found:
CSV file: file:///home/nagappan/Downloads/out.csv
24/12/08 20:58:04 INFO Executor: Finished task 0.0 in stage 7.0 (TID 9). 1735 bytes result sent to driver
24/12/08 20:58:04 INFO TaskSetManager: Finished task 0.0 in stage 7.0 (TID 9) in 128 ms on 192.168.0.103 (executor driver) (1/1)
24/12/08 20:58:04 INFO TaskSchedulerImpl: Removed TaskSet 7.0, whose tasks have all completed, from pool
24/12/08 20:58:04 INFO DAGScheduler: ResultStage 7 (show at StandaloneSparkTester.java:65) finished in 0.157 s
24/12/08 20:58:04 INFO DAGScheduler: Job 5 is finished. Cancelling potential speculative or zombie tasks for this job
24/12/08 20:58:04 INFO TaskSchedulerImpl: Killing all running tasks in stage 7: Stage finished
24/12/08 20:58:04 INFO DAGScheduler: Job 5 finished: show at StandaloneSparkTester.java:65, took 0.161751 s
24/12/08 20:58:04 INFO CodeGenerator: Code generated in 38.643298 ms
+-------------+---------------------+-------------+---------+---------------+-------------------+-------------------+------------+------------+------------+-----------------+----+
|RetailStoreID|OrganizationHierarchy|WorkstationID|InvoiceNO|BusinessDayDate| BeginDateTime| EndDateTime| OperatorID|OperatorName|CurrencyCode|TransactionTypeID|_c11|
+-------------+---------------------+-------------+---------+---------------+-------------------+-------------------+------------+------------+------------+-----------------+----+
| 9201| Kerala| 100| 77478| 2023-03-07|2023-03-07T12:25:54|2023-03-07T13:39:27| ANEESH S| 1110113| INR| OP|null|
| 9201| Kerala| 1| 103063| 2023-03-07|2023-03-07T13:34:49|2023-03-07T13:35:47|Adithya Babu| 1108822| INR| SA|null|
+-------------+---------------------+-------------+---------+---------------+-------------------+-------------------+------------+------------+------------+-----------------+----+
24/12/08 20:58:04 INFO InMemoryFileIndex: It took 2 ms to list leaf files for 1 paths.
24/12/08 20:58:04 INFO InMemoryFileIndex: It took 1 ms to list leaf files for 1 paths.
Once data frame comes, we can easily filter based on any columns by type casting to their data type and then filter the rows in the data frame.
Dataset<Row> filteredDF = csvdf.filter(col("BeginDateTime")
.cast("timestamp").gt("2023-03-07T13:00:49"));
filteredDF.show();
Output is as shown below:
24/12/10 12:36:02 INFO DAGScheduler: Job 6 finished: show at StandaloneSparkTester.java:68, took 0.074211 s
+-------------+---------------------+-------------+---------+---------------+-------------------+-------------------+------------+------------+------------+-----------------+----+
|RetailStoreID|OrganizationHierarchy|WorkstationID|InvoiceNO|BusinessDayDate| BeginDateTime| EndDateTime| OperatorID|OperatorName|CurrencyCode|TransactionTypeID|_c11|
+-------------+---------------------+-------------+---------+---------------+-------------------+-------------------+------------+------------+------------+-----------------+----+
| 9201| Kerala| 1| 103063| 2023-03-07|2023-03-07T13:34:49|2023-03-07T13:35:47|Adithya Babu| 1108822| INR| SA|null|
+-------------+---------------------+-------------+---------+---------------+-------------------+-------------------+------------+------------+------------+-----------------+----+
24/12/10 12:36:02 INFO InMemoryFileIndex: It took 1 ms to list leaf files for 1 paths.
24/12/10 12:36:02 INFO InMemoryFileIndex: It took 1 ms to list leaf files for 1 paths.
JSON Reader
Get the reader from the spark session and read the json file. Initially it will read the root level json keys in the json structure. Nested documents fields inside can be traversed and flatten the fields by using the field schema. Nested document field have data type as StructType, which will have the fields to that particular field. Using a recursive approach, we can get all the fields with link of the parent fields as columns. Then directly we can select the json read data frame to list the fields in flatten structure.
Dataset<Row> jsondf = spark
.read()
.json("/home/nagappan/Downloads/gl-sast-report.json");
System.out.println("json file read and columns are --->" + Arrays.toString(jsondf.columns()));
List<Column> colList = flattenColList(jsondf);
System.out.println("*********flattend columns********");
jsondf.select(colList.toArray(new Column[0])).show();
System.out.println("*********flattend columns********");
Below are the supporting helper methods to flatten the json structure.
public static List<Column> flattenColList(Dataset<Row> df) {
StructType schema = df.schema();
List<Column> columns = new ArrayList<>();
ConcurrentLinkedQueue<StructField> metadataFields
= new ConcurrentLinkedQueue<>(Arrays.asList(schema.fields()));
for (StructField field : metadataFields) {
System.out.println("col names -" + field.name());
if (field.dataType() instanceof StructType) {
// for (StructField nestedField : ((StructType) field.dataType()).fields()) {
// columns.add(col(field.name() + "." + nestedField.name()).alias(field.name() + "." + nestedField.name()));
// }
List<StructField> ancestors = new ArrayList<>();
columns.addAll(nestedColumns(field, ancestors));
} else {
columns.add(col(field.name()));
}
}
return columns;
}
public static List<Column> nestedColumns(StructField field, List<StructField> parentField) {
System.out.println("entered nested columns " + field + " ancestors " + parentField );
List<Column> columns = new ArrayList<>();
for (StructField nestedField : ((StructType) field.dataType()).fields()) {
System.out.println("col names -" + nestedField.name() + " parent field "+ field.name());
if (nestedField.dataType() instanceof StructType) {
List<StructField> parentCopy = new ArrayList<>(parentField); parentCopy.add(field);
columns.addAll(nestedColumns(nestedField, parentCopy));
} else {
String colName = "";
for (StructField element: parentField) {
colName += element.name() + ".";
}
colName +=field.name() + "." + nestedField.name();
System.out.println("complete field path --->" + colName);
columns.add(col(colName).alias(colName));
}
}
return columns;
}
After flattening, the column values are printed as below:
24/12/10 12:36:04 INFO Executor: Finished task 0.0 in stage 15.0 (TID 17). 1800 bytes result sent to driver
24/12/10 12:36:04 INFO TaskSetManager: Finished task 0.0 in stage 15.0 (TID 17) in 25 ms on 192.168.0.103 (executor driver) (1/1)
24/12/10 12:36:04 INFO TaskSchedulerImpl: Removed TaskSet 15.0, whose tasks have all completed, from pool
24/12/10 12:36:04 INFO DAGScheduler: ResultStage 15 (show at StandaloneSparkTester.java:80) finished in 0.032 s
24/12/10 12:36:04 INFO DAGScheduler: Job 13 is finished. Cancelling potential speculative or zombie tasks for this job
24/12/10 12:36:04 INFO TaskSchedulerImpl: Killing all running tasks in stage 15: Stage finished
24/12/10 12:36:04 INFO DAGScheduler: Job 13 finished: show at StandaloneSparkTester.java:80, took 0.036180 s
24/12/10 12:36:04 INFO CodeGenerator: Code generated in 19.847198 ms
+----------------+----------------+------------------+--------------------+-------------------------+---------------------+-------------------+---------------+-----------------+--------------------+------------------------+--------------------+-------------------+-----------+---------+-------+---------------+
|dependency_files|scan.analyzer.id|scan.analyzer.name| scan.analyzer.url|scan.analyzer.vendor.name|scan.analyzer.version| scan.end_time|scan.scanner.id|scan.scanner.name| scan.scanner.url|scan.scanner.vendor.name|scan.scanner.version| scan.start_time|scan.status|scan.type|version|vulnerabilities|
+----------------+----------------+------------------+--------------------+-------------------------+---------------------+-------------------+---------------+-----------------+--------------------+------------------------+--------------------+-------------------+-----------+---------+-------+---------------+
| []| semgrep| Semgrep|https://gitlab.co...| GitLab| 4.9.1|2023-11-25T04:10:10| semgrep| Semgrep|https://github.co...| GitLab| 1.50.0|2023-11-25T04:10:03| success| sast| 15.0.7| []|
+----------------+----------------+------------------+--------------------+-------------------------+---------------------+-------------------+---------------+-----------------+--------------------+------------------------+--------------------+-------------------+-----------+---------+-------+---------------+
*********flattend columns********
Complete source code is available in github for your reference.