Sparkのデータソース

2024/4/12に翔泳社より**Apache Spark徹底入門**を出版します！

<a href="https://www.amazon.co.jp/dp/4798182281/"><img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/87910d7c-8a63-c198-8e82-216ea7a8ee07.jpeg" width=300></a>

書籍の[サンプルノートブック](https://github.com/databricks/LearningSparkV2/blob/master/notebooks/LearningSparkv2.dbc)をウォークスルーしていきます。`Python/Chapter04/4-2 Spark Data Sources`となります。

翻訳ノートブックのリポジトリはこちら。

https://github.com/taka-yayoi/SparkSample/tree/main

ノートブックはこちら

https://github.com/taka-yayoi/SparkSample/blob/main/Chapter04/4-2%20Spark%20Data%20Sources.py

# パスの定義

```py
parquet_file = "/databricks-datasets/learning-spark-v2/flights/summary-data/parquet/2010-summary.parquet"
json_file = "/databricks-datasets/learning-spark-v2/flights/summary-data/json/*"
csv_file = "/databricks-datasets/learning-spark-v2/flights/summary-data/csv/*"
orc_file = "/databricks-datasets/learning-spark-v2/flights/summary-data/orc/*"
avro_file = "/databricks-datasets/learning-spark-v2/flights/summary-data/avro/*"
schema = "DEST_COUNTRY_NAME STRING, ORIGIN_COUNTRY_NAME STRING, count INT"
```

# Parquetデータソース

```py
df = (spark
      .read
      .format("parquet")
      .option("path", parquet_file)
      .load())
display(df)
```
![Screenshot 2024-03-27 at 13.43.57.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4b8c531f-178f-46a2-0468-dd680e45a9c4.png)

このAPIのバリエーションを用いて同じデータを読み込む別の方法。

```py
df2 = spark.read.parquet(parquet_file)
display(df2)
```
![Screenshot 2024-03-27 at 13.43.57.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4b8c531f-178f-46a2-0468-dd680e45a9c4.png)

## SQLの活用

これは _アンマネージドの_ 一時ビューを作成します。

```sql
%sql
CREATE OR REPLACE TEMPORARY VIEW us_delay_flights_tbl
    USING parquet
    OPTIONS (
      path "/databricks-datasets/definitive-guide/data/flight-data/parquet/2010-summary.parquet"
    )
```

テーブルをクエリーするためにSQLを使います。

結果は上のデータフレームに読み込んだものと同じになります。

```py
spark.sql("SELECT * FROM us_delay_flights_tbl").display()
```
![Screenshot 2024-03-27 at 13.43.57.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4b8c531f-178f-46a2-0468-dd680e45a9c4.png)

# JSONデータソース

```py
df = spark.read.format("json").option("path", json_file).load()
display(df)
```
![Screenshot 2024-03-27 at 13.46.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/3f1a1d63-92c8-4a1d-d565-4fe6eb4a0bcc.png)

```py
df2 = spark.read.json(json_file)
display(df2)
```
![Screenshot 2024-03-27 at 13.46.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/3f1a1d63-92c8-4a1d-d565-4fe6eb4a0bcc.png)

## SQLの活用

```sql
%sql
CREATE OR REPLACE TEMPORARY VIEW us_delay_flights_tbl
    USING json
    OPTIONS (
      path "/databricks-datasets/learning-spark-v2/flights/summary-data/json/*"
    )
```

```py
spark.sql("SELECT * FROM us_delay_flights_tbl").display()
```
![Screenshot 2024-03-27 at 13.46.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/3f1a1d63-92c8-4a1d-d565-4fe6eb4a0bcc.png)

# CSVデータソース

```py
df = (spark
      .read
	 .format("csv")
	 .option("header", "true")
	 .schema(schema)
	 .option("mode", "FAILFAST")  # エラーが起きたら終了
	 .option("nullValue", "")	  # nullデータを""で置換
	 .option("path", csv_file)
	 .load())
display(df)
```
![Screenshot 2024-03-27 at 13.50.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b22b5af9-b843-49b9-849b-676b6180e7d8.png)

## SQLの活用

```py
%sql
CREATE OR REPLACE TEMPORARY VIEW us_delay_flights_tbl
    USING csv
    OPTIONS (
      path "/databricks-datasets/learning-spark-v2/flights/summary-data/csv/*",
      header "true",
      inferSchema "true",
      mode "FAILFAST"
    )
```
```py
spark.sql("SELECT * FROM us_delay_flights_tbl").display()
```
![Screenshot 2024-03-27 at 13.50.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b22b5af9-b843-49b9-849b-676b6180e7d8.png)

# ORCデータソース

```py
df = (spark.read
      .format("orc")
      .option("path", orc_file)
      .load())
display(df)
```
![Screenshot 2024-03-27 at 13.52.40.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d2d8de60-2c96-bf2c-f363-23d35e420ae5.png)

## SQLの活用

```sql
%sql
CREATE OR REPLACE TEMPORARY VIEW us_delay_flights_tbl
    USING orc
    OPTIONS (
      path "/databricks-datasets/learning-spark-v2/flights/summary-data/orc/*"
    )
```
```py
spark.sql("SELECT * FROM us_delay_flights_tbl").show(10, truncate=False)
```
```
+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|United States    |Romania            |1    |
|United States    |Ireland            |264  |
|United States    |India              |69   |
|Egypt            |United States      |24   |
|Equatorial Guinea|United States      |1    |
|United States    |Singapore          |25   |
|United States    |Grenada            |54   |
|Costa Rica       |United States      |477  |
|Senegal          |United States      |29   |
|United States    |Marshall Islands   |44   |
+-----------------+-------------------+-----+
only showing top 10 rows
```

# Avroデータソース

```py
df = (spark.read
      .format("avro")
      .option("path", avro_file)
      .load())
display(df)
```
![Screenshot 2024-03-27 at 13.53.55.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5502191c-59fb-6813-afd0-b77b2d80691b.png)

## SQLの活用

```sql
%sql
CREATE OR REPLACE TEMPORARY VIEW us_delay_flights_tbl
    USING avro
    OPTIONS (
      path "/databricks-datasets/learning-spark-v2/flights/summary-data/avro/*"
    )
```
```py
spark.sql("SELECT * FROM us_delay_flights_tbl").show(10, truncate=False)
```
```
+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|United States    |Romania            |1    |
|United States    |Ireland            |264  |
|United States    |India              |69   |
|Egypt            |United States      |24   |
|Equatorial Guinea|United States      |1    |
|United States    |Singapore          |25   |
|United States    |Grenada            |54   |
|Costa Rica       |United States      |477  |
|Senegal          |United States      |29   |
|United States    |Marshall Islands   |44   |
+-----------------+-------------------+-----+
only showing top 10 rows
```

# 画像

```py
from pyspark.ml import image

image_dir = "/databricks-datasets/cctvVideos/train_images/"
images_df = spark.read.format("image").load(image_dir)
images_df.printSchema()

images_df.select("image.height", "image.width", "image.nChannels", "image.mode", "label").show(5, truncate=False)
```
```
root
 |-- image: struct (nullable = true)
 |    |-- origin: string (nullable = true)
 |    |-- height: integer (nullable = true)
 |    |-- width: integer (nullable = true)
 |    |-- nChannels: integer (nullable = true)
 |    |-- mode: integer (nullable = true)
 |    |-- data: binary (nullable = true)
 |-- label: integer (nullable = true)

+------+-----+---------+----+-----+
|height|width|nChannels|mode|label|
+------+-----+---------+----+-----+
|288   |384  |3        |16  |0    |
|288   |384  |3        |16  |1    |
|288   |384  |3        |16  |0    |
|288   |384  |3        |16  |0    |
|288   |384  |3        |16  |0    |
+------+-----+---------+----+-----+
only showing top 5 rows
```
```py
display(images_df)
```
![Screenshot 2024-03-27 at 13.55.27.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/620f4821-da66-8716-90d7-9c28e67c95fd.png)

# バイナリー

```py
path = "/databricks-datasets/learning-spark-v2/cctvVideos/train_images/"
binary_files_df = (spark
                   .read
                   .format("binaryFile")
                   .option("pathGlobFilter", "*.jpg")
                   .load(path))
binary_files_df.show(5)
```
```
+--------------------+-------------------+------+--------------------+-----+
|                path|   modificationTime|length|             content|label|
+--------------------+-------------------+------+--------------------+-----+
|dbfs:/databricks-...|2020-01-02 20:42:21| 55037|[FF D8 FF E0 00 1...|    0|
|dbfs:/databricks-...|2020-01-02 20:42:31| 54634|[FF D8 FF E0 00 1...|    1|
|dbfs:/databricks-...|2020-01-02 20:42:21| 54624|[FF D8 FF E0 00 1...|    0|
|dbfs:/databricks-...|2020-01-02 20:42:22| 54505|[FF D8 FF E0 00 1...|    0|
|dbfs:/databricks-...|2020-01-02 20:42:22| 54475|[FF D8 FF E0 00 1...|    0|
+--------------------+-------------------+------+--------------------+-----+
only showing top 5 rows
```

ディレクトリのパーティションデータディスカバリーを無視するには、`recursiveFileLookup`を`true`に設定します。

```py
binary_files_df = (spark
                   .read
                   .format("binaryFile")
                   .option("pathGlobFilter", "*.jpg")
                   .option("recursiveFileLookup", "true")
                   .load(path))
binary_files_df.show(5)
```
```
+--------------------+-------------------+------+--------------------+
|                path|   modificationTime|length|             content|
+--------------------+-------------------+------+--------------------+
|dbfs:/databricks-...|2020-01-02 20:42:21| 55037|[FF D8 FF E0 00 1...|
|dbfs:/databricks-...|2020-01-02 20:42:31| 54634|[FF D8 FF E0 00 1...|
|dbfs:/databricks-...|2020-01-02 20:42:21| 54624|[FF D8 FF E0 00 1...|
|dbfs:/databricks-...|2020-01-02 20:42:22| 54505|[FF D8 FF E0 00 1...|
|dbfs:/databricks-...|2020-01-02 20:42:22| 54475|[FF D8 FF E0 00 1...|
+--------------------+-------------------+------+--------------------+
only showing top 5 rows
```


### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

[Databricks無料トライアル](https://databricks.com/jp/try-databricks)