何の記事？
GitHubのリポジトリの中身を、備忘のためセルフ解説する記事です。
以前の記事を書いた時点からコードの更新をしたのでその記録。
リポジトリ名：dirty_data
URL：https://github.com/tsuruQQ/dirty_data.git
型が増えた
型の種類は、INT, STRING, TIMESTAMPの3種類しか想定していませんでしたが、DECIMAL, DATE, BIGINTにも対応するように修正しました。
"table_a": (
{
"id": "INT",
"name": "STRING",
"name_kana": "STRING",
"tel": "STRING",
"zip": "STRING",
"memo": "STRING",
"birthday": "TIMESTAMP"
},
1. DECIMALが現れた！
テーブル全体を見渡してもDECIMALが使われるのは1カ所のみだったので、精度とスケールは固定でも良かったのですが、一応柔軟に対応できるようにしていきます。
def generate_clean_value(col_type):
elif col_type.startsWith("DECIMAL"):
# DECIMAL(p,s)からpとsを取り出す
match = re.match(r'DECIMAL\((\d+),\s*(\d+)\)', col_type) # ★1
if match:
p = int(match.group(1)) # ★2
s = int(match.group(2)) # ★2
max_value = 10 ** (p - s) - Decimal(f'1e-{s}') # ★3
raw = random.uniform(0, float(max_value)) # ★4
return Decimal(str(raw)).quantize(Decimal(f'1.{"0" * s}'), rounding=ROUND_DOWN) # ★5
正規表現を使い、文字列DECIMAL(p, s)から精度（p）とスケール（s）を取り出します(★1)。取り出したpとsを整数に変換(★2)。最大値max_valueを計算し(★3)、0〜最大値までの間でランダムな小数を1つ生成(★4)。ランダム値をDECIMALにして、小数点以下の桁数をs桁に調整(★5)。
- ★3部分の掘り下げ
10 ** (p - s)：10の(p-s)乗
1e-{s}：10のマイナスs乗
つまり、10 ** (p - s)で整数部の最大値を出し、そこから小数部の最小単位1e-{s}を引き算することでDECIMAL(p,s)の最大値ギリギリまで攻めた値を取れるようになります。
2. DATEが現れた！
DATEは、TIMESTAMPとほぼ同様の処理としました。
elif col_type == "DATE":
return f"DATE '{datetime.today().strftime('%Y-%m-%d')}'"
def format_json_value(value, col_type):
elif col_type == "DATE":
match = re.match(r"DATE\s+'(.+)'", str(value))
return match.group(1) if match else str(value)
3. BIGINTが現れた！
BIGINTはINTの処理に仲間入りさせました。ついでにINTEGERも。
def generate_clean_value(col_type):
if col_type in ("INT", "INTEGER", "BIGINT"):
return random.randint(1000, 9999)
JSONLも必要になった
個人レベルで行う挙動確認では、INSERT文でデータ投入するだけで十分でしたが、DBとしてAmazon Athenaを使用しており、パイプラインに載せて本番相当の動作確認を行う際にJSONLでのデータ投入が必要でした。
ではJSONLを出していこうか
INSERT出力も残しつつ、JSONL形式のファイルも出力できるように修正します。sql_writer.pyに次の2つのメソッドを追加。
# jsonlフォーマット成形用
def format_json_value(value, col_type):
# JSONでは文字列化せず、型そのまま
if col_type.startswith("DECIMAL"):
return float(value)
# ...中略...
# sql、jsonl出力用
def generate_insert_sql_and_json(table_name, type_dict, category_dict, primary_keys, num_rows):
# ...中略...
for i in range(num_rows):
sql_row = [] # カラムの順番にvalueを詰めるのでリスト
json_obj = {} # key:valueの形にするので辞書
ここで悩ましかったのがgenerate_clean_value()の存在です。当初、出力形式はINSERT文だけだったので、ここでがっつりSQL用の修飾をしています。
def generate_clean_value(col_type):
# ...中略....
elif col_type == "TIMESTAMP":
return f"TIMESTAMP '{datetime.now().strftime('%Y-%m-%d 00:00:00')}'"
elif col_type == "DATE":
return f"DATE '{datetime.today().strftime('%Y-%m-%d')}'"
たとえばTIMESTAMP型ならこんな感じに整形される。
TIMESTAMP '2025-07-05 08:54:47'
これを解除するため、format_json_value()でJSONLフォーマット用の成形を行うことにしました。本当はもう少し丁寧に手術すべきなのですが、とにかく動けばいいので。
おしまい
心残りというか、気になる部分も多々ありますがこのコード改修はおそらくこの1回で終わりです。次作るとしたら一からやり直したい。