Azure Data Factory にて改行を含む区切りテキストファイルを Databricks Delta Lake コネクターにより書き込むパイプラインの実装方法

## 概要

この記事では、Azure Data Factory を使用して、下記のような改行（`\n`）を含む区切りテキスト形式のファイルを Databricks Delta Lake コネクターを介して Databricks に書き込む方法を共有します。Spark Dataframe では、カラム内に改行が含まれている場合、`multiLine`を`True`に設定することで想定通りに読み取ることができます。同様の設定は Azure Data Factory でも必要です。

```text
id,text
"1","abc"
"2","あいう
"
"3","123"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b161839a-bd00-bdc6-d84f-e2fb71f6905f.png)


Azure Data Factory でデフォルトの設定で書き込む場合、エラーが発生します。この問題を解決するために、Azure Data Factory におけるパイプライン定義の json に `additionalFormatOptions` を直接追記する方法を採用します。 

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/aaad06fc-8861-a0f9-1ad5-49ca61d5bba4.png)

以下はエラーメッセージの例です。

```text
ErrorCode=AzureDatabricksCommandError,Hit an error when running the command in Azure Databricks. Error details: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 6.0 failed 4 times, most recent failure: Lost task 0.3 in stage 6.0 (TID 9) (10.139.64.10 executor 0): org.apache.spark.SparkException: [TASK_WRITE_FAILED] Task failed while writing rows to dbfs:/user/hive/warehouse/multi_line_test.db/test.
Caused by: com.databricks.sql.io.FileReadException: Error while reading file wasbs:REDACTED_LOCAL_PART@{storage_name}.blob.core.windows.net/multiline/source.txt.
Caused by: org.apache.spark.SparkException: [MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION] Malformed records are detected in record parsing: [
Caused by: org.apache.spark.sql.catalyst.util.BadRecordException: org.apache.spark.SparkRuntimeException: [MALFORMED_CSV_RECORD] Malformed CSV record: "
Caused by: org.apache.spark.SparkRuntimeException: [MALFORMED_CSV_RECORD] Malformed CSV record: "
Caused by: org.apache.spark.SparkException: [TASK_WRITE_FAILED] Task failed while writing rows to dbfs:/user/hive/warehouse/multi_line_test.db/test.
Caused by: com.databricks.sql.io.FileReadException: Error while reading file wasbs:REDACTED_LOCAL_PART@{storage_name}.blob.core.windows.net/multiline/source.txt.
Caused by: org.apache.spark.SparkException: [MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION] Malformed records are detected in record parsing: [
Caused by: org.apache.spark.sql.catalyst.util.BadRecordException: org.apache.spark.SparkRuntimeException: [MALFORMED_CSV_RECORD] Malformed CSV record: "
Caused by: org.apache.spark.SparkRuntimeException: [MALFORMED_CSV_RECORD] Malformed CSV record: ".
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/45fc9d54-171b-dc3c-f401-7bd01327be2d.png)

``

Databricks における`multiLine`について、下記の記事において紹介しています。

[Databricks ( Spark ) にて 引用符 ( sep ) 内に 区切り文字 ( lineSep ) が含まれる CSV ファイルを取り込む際には `multiLine` を `true` に設定すべき #Python - Qiita](https://qiita.com/manabian/items/0d9c56b927e73268a086)

## エラーとなるコードと対応策
### 事前準備

改行を含む区切りテキストファイルを Azure Storage に配置します。

```text
id,text
"1","abc"
"2","あいう
"
"3","123"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/750f55fb-dafe-d46c-7a2c-b747313a8f22.png)


Databricks からストレージのファイルを想定通りに参照できることを確認します。

```python
storage_name = ""  # Azure Storage名
container_name = ""  # Azure Storage のコンテナー名

df = (
    spark.read.format("csv")
    .option("header", True)
    .option("multiLine", True)
    .load(f"wasbs://{container_name}@{storage_name}.blob.core.windows.net/multiline")
)

df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b1a4665d-c243-0e85-84c8-9aeece70da2d.png)


Databricks で書き込み先の Hive metastore のテーブルを作成します。

```python
%sql
CREATE SCHEMA IF NOT EXISTS multi_line_test;
CREATE OR REPLACE TABLE multi_line_test.test (
  id string,
  text string
);
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b0f0042f-5ea9-dc5c-c611-64e8c6931fba.png)

### エラー事象を確認

Azure Storage から Databricks へ書き込むパイプラインを作成して実行します。すると、下記のようなエラーが発生します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1be085b6-497c-cf9a-328d-85152d3f0683.png)


```text
ErrorCode=AzureDatabricksCommandError,Hit an error when running the command in Azure Databricks. Error details: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 6.0 failed 4 times, most recent failure: Lost task 0.3 in stage 6.0 (TID 9) (10.139.64.10 executor 0): org.apache.spark.SparkException: [TASK_WRITE_FAILED] Task failed while writing rows to dbfs:/user/hive/warehouse/multi_line_test.db/test.
Caused by: com.databricks.sql.io.FileReadException: Error while reading file wasbs:REDACTED_LOCAL_PART@{storage_name}.blob.core.windows.net/multiline/source.txt.
Caused by: org.apache.spark.SparkException: [MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION] Malformed records are detected in record parsing: [
Caused by: org.apache.spark.sql.catalyst.util.BadRecordException: org.apache.spark.SparkRuntimeException: [MALFORMED_CSV_RECORD] Malformed CSV record: "
Caused by: org.apache.spark.SparkRuntimeException: [MALFORMED_CSV_RECORD] Malformed CSV record: "
Caused by: org.apache.spark.SparkException: [TASK_WRITE_FAILED] Task failed while writing rows to dbfs:/user/hive/warehouse/multi_line_test.db/test.
Caused by: com.databricks.sql.io.FileReadException: Error while reading file wasbs:REDACTED_LOCAL_PART@{storage_name}.blob.core.windows.net/multiline/source.txt.
Caused by: org.apache.spark.SparkException: [MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION] Malformed records are detected in record parsing: [
Caused by: org.apache.spark.sql.catalyst.util.BadRecordException: org.apache.spark.SparkRuntimeException: [MALFORMED_CSV_RECORD] Malformed CSV record: "
Caused by: org.apache.spark.SparkRuntimeException: [MALFORMED_CSV_RECORD] Malformed CSV record: ".
```

### エラーへの対応策を実施

Azure Data Factory にて、パイプライン右上に表示されている`このソースの json コード表現を表示する`をクリックして、定義の json を表示します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4fba4eb9-26b0-9d80-59d7-6fce86c17a2a.png)

次に、sink の項目を探し、importSettings に `additionalFormatOptions` を追加します。

```json
"additionalFormatOptions": {
    "multiline": true
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a00b676a-9cf5-10c7-08cb-c40faca8bb38.png)


設定前後を比較すると、次のようになります。

```json
"sink": {
    "type": "AzureDatabricksDeltaLakeSink",
    "importSettings": {
        "type": "AzureDatabricksDeltaLakeImportCommand"
    }
},
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4e4a301b-3852-d3e1-80d9-2064b0566a42.png)


```json
"sink": {
    "type": "AzureDatabricksDeltaLakeSink",
    "importSettings": {
        "type": "AzureDatabricksDeltaLakeImportCommand",
        "additionalFormatOptions": {
            "multiline": true
        }
    }
},
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/de730766-c179-186d-a0ed-c6d73eade555.png)


パイプラインを実行し、正常終了することを確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/70312bd6-d359-4cda-7418-210a3c86b960.png)

最後に、Databricks でテーブルに書き込まれていることを確認します。

```python
df = spark.table("multi_line_test.test")

print(df.collect())
df.display()
```

```python
[
    Row(id="1", text="abc"),
    Row(id="2", text="あいう\n"),
    Row(id="3", text="123"),
]
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e3d43bf7-202d-961e-b707-85c6e42695be.png)