Spark Ozone Integration in CDP
In this tutorial, we will learn how to create Apache Ozone volumes, buckets, and keys. After that, we will see how we can access Apache Ozone data in Apache Spark.
Ozone
- Create the volume with the name vol1 in Apache Ozone.
# ozone sh volume create /vol1
21/08/25 06:23:27 INFO rpc.RpcClient: Creating Volume: vol1, with root as owner. - Create the bucket with the name bucket1 under vol1.
# ozone sh bucket create /vol1/bucket1
21/08/25 06:24:09 INFO rpc.RpcClient: Creating Bucket: vol1/bucket1, with Versioning false and Storage Type set to DISK and Encryption set to false - Create the employee.csv file to upload to Ozone.
# vi /tmp/employee.csv
id,name,age
1,Ranga,33
2,Nishanth,4
3,Raja,60 - Upload the employee.csv file to Ozone
# ozone sh key put /vol1/bucket1/employee.csv /tmp/employee.csv - Add the fs.o3fs.impl property to core-site.xml
- Go to Cloudera Manager > HDFS > Configuration > search for core-site.xml > Cluster-wide Advanced Configuration Snippet (Safety Valve) for core-site.xml
<property>
<name>fs.o3fs.impl</name>
<value>org.apache.hadoop.fs.ozone.OzoneFileSystem</value>
</property>
- Go to Cloudera Manager > HDFS > Configuration > search for core-site.xml > Cluster-wide Advanced Configuration Snippet (Safety Valve) for core-site.xml
- Display the files created earlier using 'hdfs' command.
Note: Before running the following command, update the om-host.example.com value.hdfs dfs -ls o3fs://bucket1.vol1.om-host.example.com/
Spark
- Launch spark-shell
spark spark-shell - Run the following command to print the employee.csv file content.
Note: Update the omHost value.scala> val omHost="om.host.example.com"
scala> val df=spark.read.option("header", "true").option("inferSchema", "true").csv(s"o3fs://bucket1.vol1.${omHost}/employee.csv")
scala> df.show()
+---+--------+---+
| id| name|age|
+---+--------+---+
| 1| Ranga| 33|
| 2|Nishanth| 4|
| 3| Raja| 60|
+---+--------+---+
Kerberized environment
Pre-requisites:
- Create a user and provide proper Ranger permissions to create Ozone volume and buckets, etc.
- kinit with the user
Steps:
- Create Ozone volumes, buckets, and keys mentioned in Ozone section.
- Launch spark-shell
- Replace the KEY_TAB, PRINCIPAL, and om.host.example.com in spark-shell
Note: In a Kerberized environment, mandatorily, we need to specify the spark.yarn.access.hadoopFileSystems configuration, otherwise, it will display the following error:spark-shell \
--keytab ${KEY_TAB} \
--principal ${PRINCIPAL} \
--conf spark.yarn.access.hadoopFileSystems=o3fs://bucket1.vol1.om.host.example.com:9862java.io.IOException: org.apache.hadoop.security.AccessControlException: Client cannot authenticate via:[TOKEN, KERBEROS] - Run the following command to print the employee.csv file content.
Note: Update the omHost value.scala> val omHost="om.host.example.com"
scala> val df=spark.read.option("header", "true").option("inferSchema", "true").csv(s"o3fs://bucket1.vol1.${omHost}/employee.csv")
scala> df.show()
+---+--------+---+
| id| name|age|
+---+--------+---+
| 1| Ranga| 33|
| 2|Nishanth| 4|
| 3| Raja| 60|
+---+--------+---+
scala> val age30DF = df.filter(df("age") > 30)
scala> val outputPath = s"o3fs://bucket1.vol1.${omHost}/employee_age30.csv"
scala> age30DF.write.option("header", "true").mode("overwrite").csv(outputPath)
scala> val df2=spark.read.option("header", "true").option("inferSchema", "true").csv(outputPath)
scala> df2.show()
+---+-----+---+
| id| name|age|
+---+-----+---+
| 1|Ranga| 33|
| 3| Raja| 60|
+---+-----+---+Note: If you get the java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.ozone.OzoneFileSystem not found, add the /opt/cloudera/parcels/CDH/jars/hadoop-ozone-filesystem-hadoop3-*.jar to spark class path using --jars option.
Thanks for reading this article. If you liked this article, you can give kudos.
