Spark-xml を使ったXML データの読み取り、XML ファイル作成　No.50

例) wget https://repo1.maven.org/maven2/com/databricks/spark-xml_2.12/0.14.0/spark-xml_2.12-0.14.0.jar

例) wget https://repo1.maven.org/maven2/com/sun/xml/bind/jaxb-core/3.0.0-M2/jaxb-core-3.0.0-M2.jar

(例)　wget https://github.com/databricks/spark-xml/raw/master/src/test/resources/books.xml

hdfs dfs -put ./books.xml appbox

1. spark-xml の jar ファイル及び、XML document と Java Object をマッピングする JAXB の jar を指定して pyspark を起動します。
2. DataFrame を作成するために pyspark.sql.SparkSession をインポートし、SparkSession を作成します。
3. books.xml ファイルからデータ部分読み取り DataFrame に入れます。
4. さらに _id および author の情報のみ抽出し、新しい xml ファイル “newbooks43.xml” に保存し、中身を確認します。
5. 読み込んだデータを parquet ファイルに保存します。保存後、hdfs dfs -ls のコマンドで保存した parquet ファイルを確認します。
6. Spark UI で Spark Job の実行履歴を確認します。

$ pyspark --jars spark-xml_2.12-0.14.0.jar,jaxb-core-3.0.0-M2.jar
Python 3.8.10 (default, Sep 28 2021, 16:10:42) 
[GCC 9.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
2021-10-19 22:26:57,756 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 3.1.2
      /_/

Using Python version 3.8.10 (default, Sep 28 2021 16:10:42)
Spark context Web UI available at http://xxxxx:4040
Spark context available as 'sc' (master = local[*], app id = local-1634650023016).
SparkSession available as 'spark'.
>>> 

>>> from pyspark.sql import SparkSession
>>> spark = SparkSession.builder.getOrCreate()

>>> df = spark.read.format('xml').options(rowTag='book').load('appbox/books.xml')
>>> df.show()
+-----+--------------------+--------------------+---------------+-----+------------+--------------------+
|  _id|              author|         description|          genre|price|publish_date|               title|
+-----+--------------------+--------------------+---------------+-----+------------+--------------------+
|bk101|Gambardella, Matthew|            An in...|       Computer|44.95|  2000-10-01|XML Developer's G...|
|bk102|          Ralls, Kim|A former architec...|        Fantasy| 5.95|  2000-12-16|       Midnight Rain|
xxx (省略) xxxx
|bk111|        O'Brien, Tim|The Microsoft MSX...|       Computer|36.95|  2000-12-01|MSXML3: A Compreh...|
|bk112|         Galos, Mike|Microsoft Visual ...|       Computer|49.95|  2001-04-16|Visual Studio 7: ...|
+-----+--------------------+--------------------+---------------+-----+------------+--------------------+

>>> df.select("author", "_id").write.format('xml').options(rowTag='book', rootTag='books').save('appbox/newbooks43.xml')
>>> df43 = spark.read.format('xml').options(rowTag='book').load('appbox/newbooks43.xml')
>>> df43.show()
+-----+--------------------+
|  _id|              author|
+-----+--------------------+
|bk101|Gambardella, Matthew|
|bk102|          Ralls, Kim|
xxx (省略) xxxx
|bk111|        O'Brien, Tim|
|bk112|         Galos, Mike|
+-----+--------------------+

1. spark-xml の jar ファイルを指定して spark-shell を起動します。
2. 上記 3 同様に books.xml ファイルからデータ部分読み取り、_id および author の情報のみ抽出します。

$ $SPARK_HOME/bin/spark-shell --packages com.databricks:spark-xml_2.12:0.14.0
:: loading settings :: url = jar:file:/home/hadoop/spark/jars/ivy-2.4.0.jar!/org/apache/ivy/core/settings/ivysettings.xml
Ivy Default Cache set to: /home/hadoop/.ivy2/cache
The jars for the packages stored in: /home/hadoop/.ivy2/jars
com.databricks#spark-xml_2.12 added as a dependency
:: resolving dependencies :: org.apache.spark#spark-submit-parent-06044b31-f2eb-4b01-b819-f3ac50700450;1.0
	confs: [default]
	found com.databricks#spark-xml_2.12;0.14.0 in central
	found commons-io#commons-io;2.8.0 in central
	found org.glassfish.jaxb#txw2;2.3.4 in central
	found org.apache.ws.xmlschema#xmlschema-core;2.2.5 in central
:: resolution report :: resolve 584ms :: artifacts dl 20ms
	:: modules in use:
	com.databricks#spark-xml_2.12;0.14.0 from central in [default]
	commons-io#commons-io;2.8.0 from central in [default]
	org.apache.ws.xmlschema#xmlschema-core;2.2.5 from central in [default]
	org.glassfish.jaxb#txw2;2.3.4 from central in [default]
	---------------------------------------------------------------------
	|                  |            modules            ||   artifacts   |
	|       conf       | number| search|dwnlded|evicted|| number|dwnlded|
	---------------------------------------------------------------------
	|      default     |   4   |   0   |   0   |   0   ||   4   |   0   |
	---------------------------------------------------------------------
:: retrieving :: org.apache.spark#spark-submit-parent-xx44b31-f2eb-4b01-b819-f3ac50700xxx
	confs: [default]
	0 artifacts copied, 4 already retrieved (0kB/24ms)
2022-05-15 21:24:33,501 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context Web UI available at http://xxxx:4040
Spark context available as 'sc' (master = local[*], app id = local-xx526174870xx).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.1.2
      /_/
         
Using Scala version 2.12.10 (OpenJDK 64-Bit Server VM, Java 1.8.0_292)
Type in expressions to have them evaluated.
Type :help for more information.

scala> 

scala> import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SparkSession

scala> import com.databricks.spark.xml._
import com.databricks.spark.xml._

scala> val spark=SparkSession.builder().getOrCreate()
spark: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@56cd8903

scala> val df=spark.read.option("rowTag","book").xml("appbox/books.xml")
df: org.apache.spark.sql.DataFrame = [_id: string, author: string ... 5 more fields]

scala> val selectedData = df.select("author","_id")
selectedData: org.apache.spark.sql.DataFrame = [author: string, _id: string]

scala> selectedData.show()
+--------------------+-----+
|              author|  _id|
+--------------------+-----+
|Gambardella, Matthew|bk101|
|          Ralls, Kim|bk102|
xxx (省略) xxxx
|        O'Brien, Tim|bk111|
|         Galos, Mike|bk112|
+--------------------+-----+