概要
この記事では、Databricks を使用して Azure SQL Database の CHAR 型カラムから末尾の空白スペースを自動的に削除する RTRIM 処理の方法を紹介します。Azure SQL Database では、CHAR 型カラムのデータが指定されたバイト数に満たない場合、残りのスペースが空白で埋められます。Databricks でのデータ操作では、これらの空白スペースはしばしば不要です。そのため、Azure SQL Database のテーブル定義情報を基に、不要なスペースを削除する RTRIM 処理を行う方法を説明します。
以前の記事 Databricks にて jdbc 経由で Azure SQL Database の CHAR 型のカラムに読み込みと書き込みを実施した際のスペースの取り扱いに関する調査結果 #Python - Qiita で触れた内容を踏まえています。
検証コードと実行結果
Azure SQL Database への接続情報を定義
以下のコードは、Azure SQL Database への接続情報を定義しています。サーバー名、データベース名、ユーザー名、パスワードを指定し、Databricks のシークレットスコープを使用して認証情報を安全に管理しています。
server = "ads-01" ## サーバー名
database = "ads-01" ## データベース名
username = dbutils.secrets.get(scope="qiita", key="sql_authentificate_user") ## Azure SQL DB のユーザー名
password = dbutils.secrets.get(scope="qiita", key="sql_authentificate_password") ## Azure SQL DB のパスワード
Azure SQL Dataabse にてテーブルを作成しデータを挿入
次のステップでは、Azure SQL Database 上にテーブルを作成し、テストデータを挿入しています。CHAR 型と VARCHAR 型のカラムを含むテーブルを定義し、サンプルデータを挿入後、Databricks でデータを読み込んで表示しています
# CREATE TABLE の実行
sql = """
IF OBJECT_ID('dbo.char_cols_test', 'U') IS NOT NULL
DROP TABLE IF EXISTS dbo.char_cols_test;
CREATE TABLE dbo.char_cols_test
(
char_col_01 CHAR(10),
varchar_col_01 VARCHAR(10),
char_col_02 CHAR(10),
char_col_03 CHAR(10),
char_col_04 CHAR(10),
varchar_col_02 VARCHAR(10)
);
INSERT INTO dbo.char_cols_test
VALUES
('123', '1234', '1234', '12345', '123456', '1234567890');
"""
df = (
spark.read.format("sqlserver")
.option("host", f"{server}.database.windows.net")
.option("user", username)
.option("password", password)
.option("database", database)
.option("prepareQuery", sql)
.option("dbTable", "dbo.char_cols_test")
.load()
)
print(df.collect())
df.display()
Azure SQL Database からテーブル定義の情報を取得して RTRIM 処理の変数を定義
このコードブロックでは、Azure SQL Database からテーブル定義の情報を取得し、RTRIM 処理を行うための変数を定義しています。CHAR 型のカラムのみを抽出し、それらのカラムに対して rtrim 関数を適用することで、末尾の空白を削除します。
schema_and_table_name = "dbo.char_cols_test"
tgt_col_name = "column_name"
get_cols_sql = f"""
SELECT
s.name AS schema_name,
t.name AS table_name,
c.name AS {tgt_col_name}
FROM sys.columns c
INNER JOIN sys.tables t ON c.object_id = t.object_id
INNER JOIN sys.schemas s ON t.schema_id = s.schema_id
WHERE
concat(s.name, '.', t.name) = N'{schema_and_table_name}'
AND type_name(c.user_type_id) = 'CHAR'
"""
df = (
spark.read.format("sqlserver")
.option("host", f"{server}.database.windows.net")
.option("user", username)
.option("password", password)
.option("database", database)
.option("query", get_cols_sql)
.load()
)
df.display()
tgt_data_type_cols = []
df_data = df.select(tgt_col_name).collect()
# df_data[0][tgt_col_name]
for r_data in df_data:
tgt_data_type_cols.append(r_data[tgt_col_name])
tgt_data_type_cols
from pyspark.sql.functions import expr
rtirm_cols = {}
for tgt_c in tgt_data_type_cols:
rtirm_cols[tgt_c] = expr(f"rtrim({tgt_c})")
Azure SQL Database からテーブルのデータを取得して RTRIM 処理を実行
以下のコードは、Azure SQL Database からテーブルのデータを取得し、定義した RTRIM 処理を実行しています。この処理により、CHAR 型カラムの末尾にある不要な空白スペースが削除されます。
src_df = (
spark.read.format("sqlserver")
.option("host", f"{server}.database.windows.net")
.option("user", username)
.option("password", password)
.option("database", database)
.option("dbTable", "dbo.char_cols_test")
.load()
)
src_df = src_df.withColumns(rtirm_cols)
print(src_df.collect())
src_df.display()
事後処理
DROP コマンドを使用して削除します。
# CREATE TABLE の実行
sql = """
IF OBJECT_ID('dbo.char_cols_test', 'U') IS NOT NULL
DROP TABLE IF EXISTS dbo.char_cols_test;
"""
df = (
spark.read.format("sqlserver")
.option("host", f"{server}.database.windows.net")
.option("user", username)
.option("password", password)
.option("database", database)
.option("prepareQuery", sql)
.option("dbTable", "(SELECT 1 AS test) src")
.load()
)
