Databricks ( Spark ) にて固定長テキストファイル（fixed-length text files）をソースとした場合における Spark データフレームの作成方法

## 概要
Databricks ( Spark ) にて 固定長テキストファイル（fixed-length text files）をソースとした場合のデータフレーム作成方法を共有します。

固定長テキストファイル（fixed-length text files）とは、下記のように１レコードの長さがスペースなどで固定（下記例では、全部で80字で固定）されているファイルです。

```text
1                   user_aaa            user_aaa@test.com   53.5                180                 
2                   user_bbb            user_bbb@test.com   53.5                180                 
3                   user_ccc            user_ccc@test.com   53.5                180                 

```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/50a3bc40-1046-121c-106e-e27fb39381ad.png)

単一列で取り込み後、`substring` 関数と `trim` 関数により新規列と追加することで、想定のデータフレームとなります。

```python
from pyspark.sql.functions import substring, trim
 
 
def split_fixed_width_cols(
    df,
    col_num,
    col_str_width,
    base_col_name_prefix="_c",
    should_drop_first_col=True,
):
    first_col_name = df.columns[0]
    counter = 1
    for tgt_col_num in range(col_num):
        new_col_name = f"{base_col_name_prefix}{str(counter)}"
        pos = tgt_col_num * col_str_width
        df = df.withColumn(
            new_col_name, trim(substring(df[first_col_name], pos, col_str_width))
        )
        counter += 1
    if should_drop_first_col:
        df = df.drop(first_col_name)
    return df
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f77ba714-4bf6-1128-f906-9d5add89717d.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0895d615-a62a-e894-36ab-d96a8d49b05e.png)

固定長テキストファイルのパターンとしては、文字数ではなく、バイト数が固定のファイルがあります。下記図がそのサンプルであり、1 行目の最初の値が `あ`というマルチバイトであるため、その後のスペースが 2 行目や 3 行目より 1つ少なくなっております。その対応方法としては、バイナリーにエンコーディング後、バイト単位でバイナリーから値をを取得して、デコードすることで対応できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3c92a659-2349-6b9c-8b0d-f9506f617684.png)

ただし、PySpark にて、`shift_jis` のバイト単位での固定ファイルを区切る場合には、対応しているエンコーディングが限られていることに注意してください。検証した際にはコードがに動作しましたが、想定外の動作をする可能性があります。また、Pandas API on Spark も、2022 年 11 月 7 日時点にて、サポート外のようです。性能に懸念がありますが、 Pandas で実装する方針がよさそうです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/255c9bf5-f9d3-3286-88c5-726e21b4db33.png)

引用元：[pyspark.sql.functions.encode — PySpark 3.1.3 documentation (apache.org)](https://spark.apache.org/docs/3.1.3/api/python/reference/api/pyspark.sql.functions.encode.html)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/23d847f5-75bb-d5bc-1676-94a1857db8ac.png)

引用元：[pyspark.pandas.Series.str.encode — PySpark 3.3.1 documentation (apache.org)](https://spark.apache.org/docs/3.3.1/api/python/reference/pyspark.pandas/api/pyspark.pandas.Series.str.encode.html#pyspark.pandas.Series.str.encode)

Windows の機種依存文字がソースに含まれる場合には、`shift_jis` ではなく、`cp932` を指定したほうがいい場合があります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/eb397084-9818-fb80-9a3d-f59914c0fa3c.png)


## 実行例手順
### 1. すべてのカラムが共通の長さである場合の実行例
```python
import inspect
from pyspark.sql import DataFrame, SparkSession
from pyspark.dbutils import DBUtils
 
 
def put_files_to_storage(
    path,
    content,
    is_overwrite=True,
):
    # 最初の行と最後の行を削除
    content = inspect.cleandoc(content)
 
    spark = SparkSession.getActiveSession()
    dbutils = DBUtils(spark)
    return dbutils.fs.put(path, content, is_overwrite)

# 検証用ファイルの配置
path = "dbfs:/test/test.txt"
 
csv_data_rows = """
1                   user_aaa            user_aaa@test.com   53.5                180                 
2                   user_bbb            user_bbb@test.com   53.5                180                 
3                   user_ccc            user_ccc@test.com   53.5                180                 
"""
 
put_files_to_storage(path, csv_data_rows)
 
# ファイルを確認
file_content = dbutils.fs.head(path)
print(file_content)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0e889247-6a75-30a7-ea7c-7a698d41d9be.png)

```python
# データフレームを作成
df = spark.read.format("text").load(path)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/99d86b9f-bca0-f7f8-a42d-e602f65b6cb8.png)

```python
from pyspark.sql.functions import substring, trim
 
 
def split_fixed_width_cols(
    df,
    col_num,
    col_str_width,
    base_col_name_prefix="_c",
    should_drop_first_col=True,
):
    first_col_name = df.columns[0]
    counter = 1
    for tgt_col_num in range(col_num):
        new_col_name = f"{base_col_name_prefix}{str(counter)}"
        pos = tgt_col_num * col_str_width
        df = df.withColumn(
            new_col_name, trim(substring(df[first_col_name], pos, col_str_width))
        )
        counter += 1
    if should_drop_first_col:
        df = df.drop(first_col_name)
    return df
```

```python
col_num = 5
col_str_width = 20
 
df_2 = split_fixed_width_cols(df, col_num, col_str_width)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a230df2f-cd0b-176a-5415-8f68a71ae53e.png)


```python
df_2.display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/49f38f85-e3b9-054b-2494-7549f968b41a.png)

### 2. カラムごとに長さが異なる場合の実行例

```python
import inspect
from pyspark.sql import DataFrame, SparkSession
from pyspark.dbutils import DBUtils
 
 
def put_files_to_storage(
    path,
    content,
    is_overwrite=True,
):
    # 最初の行と最後の行を削除
    content = inspect.cleandoc(content)
 
    spark = SparkSession.getActiveSession()
    dbutils = DBUtils(spark)
    return dbutils.fs.put(path, content, is_overwrite)

# 検証用ファイルの配置
path = "dbfs:/test/test.csv"
 
csv_data_rows = """
1    user_aaa            user_aaa@test.com   53.5      180  
2    user_bbb            user_bbb@test.com   53.5      180  
3    user_ccc            user_ccc@test.com   53.5      180  
"""

_ = put_files_to_storage(path, csv_data_rows)
 
```python
# ファイルを確認
file_content = dbutils.fs.head(path)
print(file_content)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/93f5b666-47bb-5096-546c-5220ad7c3ef9.png)


```python
# データフレームを作成
df = spark.read.format("text").load(path)
df.display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ec6cebab-4310-2734-2975-4f241fed3c4b.png)


```python

from pyspark.sql.functions import substring, trim
 
 
def split_fixed_width_cols_with_col_width(
    df,
    col_str_widths,
    base_col_name_prefix="_c",
    should_drop_first_col=True,
):
    first_col_name = df.columns[0]
    pos = 0
    
    counter = 1
    for tgt_col_str_width in col_str_widths:
        new_col_name = f"{base_col_name_prefix}{str(counter)}"
        df = df.withColumn(
            new_col_name, trim(substring(df[first_col_name], pos, tgt_col_str_width))
        )
        pos += tgt_col_str_width
        counter += 1
    if should_drop_first_col:
        df = df.drop(first_col_name)
    return df
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e04845ca-74c7-b6f2-e580-cb3b43848511.png)


```python
col_str_widths = [5, 20, 20, 10, 5]
 
df_3 = split_fixed_width_cols_with_col_width(df, col_str_widths)

```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/53d7fa72-2701-2083-833c-68201589c6eb.png)


```python
df_3.display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/70a23bc0-288a-4edb-a28b-66d148ade939.png)

### 3. すべてのカラムが共通のバイト数である場合の実行例（文字コードが`shift_jis`）

```python
import inspect
from pyspark.sql import DataFrame, SparkSession
from pyspark.dbutils import DBUtils
 
 
def put_files_to_storage(
    path,
    content,
    is_overwrite=True,
):
    # 最初の行と最後の行を削除
    content = inspect.cleandoc(content)
 
    spark = SparkSession.getActiveSession()
    dbutils = DBUtils(spark)
    return dbutils.fs.put(path, content, is_overwrite)
  
# 検証用ファイルの配置
path = "dbfs:/test/test.txt"
 
csv_data_rows = """
あ        user_あ   a         
2         user_b    b         
3         user_c    c         
"""
 
_ = put_files_to_storage(path, csv_data_rows,)
 
# ファイルを確認
file_content = dbutils.fs.head(path)
print(file_content)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/64737e64-73cf-4f20-f3c3-79ac4b44bb96.png)


```python
# データフレームを作成
df = spark.read.format("text").load(path)
df.display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3db1881a-5481-8d1a-7b4a-9addba0c9a91.png)

```python
from pyspark.sql.functions import substring, trim, encode, decode
 
def split_fixed_byte_cols_with_encoding(
    df,
    col_num,
    col_byte,
    encoding = 'shift_jis',
    errors = 'ignore',
    base_col_name_prefix="_c",
    should_drop_first_col=True,
):
    first_col_name = df.columns[0]
    counter = 1
    pdf = df.toPandas()
    for tgt_col_num in range(col_num):
        start = tgt_col_num * col_byte
        stop = counter *  col_byte
        new_col_name = f"{base_col_name_prefix}{str(counter)}"
        pdf[new_col_name] = pdf[first_col_name].str.encode(encoding)
        pdf[new_col_name] = pdf[new_col_name].str.slice(start, stop)
        pdf[new_col_name] = pdf[new_col_name].str.decode(encoding, errors)
        counter += 1
 
    df = spark.createDataFrame(pdf)
 
    counter = 1
    for tgt_col_num in range(col_num):
        tgt_col_name = f"{base_col_name_prefix}{str(counter)}"
        df = df.withColumn(
            tgt_col_name, trim(df[tgt_col_name])
        )
        counter += 1
    
    if should_drop_first_col:
        df = df.drop(first_col_name)
    return df
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/baedf9ab-86bc-4b3f-8557-91338c1f2f97.png)


```python
col_num = 3
col_byte = 10
 
df_2 = split_fixed_byte_cols_with_encoding(df, col_num, col_byte)
df_2.display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d4c74949-1ea1-db46-7bf8-61ed23902482.png)

下記のように PySpark での動作を確認済みだが、前述のようにサポート外。

```python
from pyspark.sql.functions import substring, trim, encode, decode
 
def split_fixed_byte_cols(
    df,
    col_num,
    col_byte,
    charset = 'shift_jis',
    errors = 'ignore',
    base_col_name_prefix="_c",
    should_drop_first_col=True,
):
    first_col_name = df.columns[0]
    counter = 1
    df = df.withColumn(first_col_name, encode(df[first_col_name], charset))
    for tgt_col_num in range(col_num):
        new_col_name = f"{base_col_name_prefix}{str(counter)}"
        start = 1 + (tgt_col_num * col_byte)
 
        df = df.withColumn(new_col_name, substring(df[first_col_name], start, col_byte))
        df = df.withColumn(new_col_name, decode(df[new_col_name], charset))
        counter += 1
 
    df = df.withColumn(first_col_name, decode(df[first_col_name], charset))
    if should_drop_first_col:
        df = df.drop(first_col_name)
    return df

col_num = 3
col_byte = 10
df_2 = split_fixed_byte_cols(df, col_num, col_byte)
df_2.display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4ddfeeba-2b89-114e-f9a8-090b81243d70.png)


### 4. カラムごとにバイト数が異なる場合の実行例（文字コードが`shift_jis`）

```python
import inspect
from pyspark.sql import DataFrame, SparkSession
from pyspark.dbutils import DBUtils
 
 
def put_files_to_storage(
    path,
    content,
    is_overwrite=True,
):
    # 最初の行と最後の行を削除
    content = inspect.cleandoc(content)
 
    spark = SparkSession.getActiveSession()
    dbutils = DBUtils(spark)
    return dbutils.fs.put(path, content, is_overwrite)
  
# 検証用ファイルの配置
path = "dbfs:/test/test.txt"
 
csv_data_rows = """
あ   user_あ   a  
2    user_b    b  
3    user_c    c  
"""
 
_ = put_files_to_storage(path, csv_data_rows,)
 
# ファイルを確認
file_content = dbutils.fs.head(path)
print(file_content)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/010c5d74-625c-1447-ec68-5817c94c4bf4.png)


```python
# データフレームを作成
df = spark.read.format("text").load(path)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/30fa1bab-91c2-aaf6-3a19-bc4d8861bf2e.png)

```python
from pyspark.sql.functions import substring, trim, encode, decode

def split_fixed_bytes_cols_with_encoding(
    df,
    col_bytes,
    encoding = 'shift_jis',
    errors = 'ignore',
    base_col_name_prefix="_c",
    should_drop_first_col=True,
):
    first_col_name = df.columns[0]
    start = 0
    stop = 0 
    
    pdf = df.toPandas()
    counter = 1
    for tgt_col_byte in col_bytes:
        stop += tgt_col_byte
        new_col_name = f"{base_col_name_prefix}{str(counter)}"
        pdf[new_col_name] = pdf[first_col_name].str.encode(encoding)
        pdf[new_col_name] = pdf[new_col_name].str.slice(start, stop)
        pdf[new_col_name] = pdf[new_col_name].str.decode(encoding, errors)
        start += tgt_col_byte
        counter += 1

    df = spark.createDataFrame(pdf)
    counter = 1
    for tgt_col_byte in col_bytes:
        tgt_col_name = f"{base_col_name_prefix}{str(counter)}"
        df = df.withColumn(
            tgt_col_name, trim(df[tgt_col_name])
        )
        counter += 1
    

    if should_drop_first_col:
        df = df.drop(first_col_name)
    return df
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6a2b27f1-2d8b-e35a-141a-3aec45c2e320.png)


```python
col_bytes = [5, 10, 3]
 
df_2 = split_fixed_bytes_cols_with_encoding(df, col_bytes)
df_2.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/024839d6-0382-5e9b-02c7-40ea315d7c2e.png)

### 事後処理

```python
# 本手順で作成したリソースを削除
dbutils.fs.rm(path, True)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5e05a99e-8827-caf0-b6a3-584f67b6abac.png)