nik/newdwh2021
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #502 feature-NEWDWH2021-1873-fix-file-row-cnt into develop

Browse Source
This commit is contained in:
下田雅人 2025-06-13 12:16:58 +09:00
commit 0d16f26f5c
1 changed files with 26 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
ecs/dataimport/dataimport/main.py
View File

@ -382,14 +382,14 @@ def import_data_with_commit_per_row(
print( print(
f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {WARNING} W-MAIN-01 {index} ロードスキーマ登録時にエラーが発生しました {line} {e}') f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {WARNING} W-MAIN-01 {index} ロードスキーマ登録時にエラーが発生しました {line} {e}')
return { return {
"counts": { "counts": {
"process": process_count, "process": process_count,
"normal": normal_count, "normal": normal_count,
"warning": warning_count "warning": warning_count
}, },
"warning_info": warning_info "warning_info": warning_info
} }
def import_data_with_bulk( def import_data_with_bulk(
@ -430,6 +430,8 @@ def import_data_with_bulk(
process_count = 0 # 処理件数カウンタ process_count = 0 # 処理件数カウンタ
normal_count = 0 # 正常終了件数カウンタ normal_count = 0 # 正常終了件数カウンタ
warning_count = 0 # ワーニング終了件数カウンター
warning_info = '' # ワーニング情報
load_schema_name = settings_list[SETTINGS_ITEM["loadSchemaName"]] load_schema_name = settings_list[SETTINGS_ITEM["loadSchemaName"]]
has_header = settings_list[SETTINGS_ITEM["headerFlag"]] or int( has_header = settings_list[SETTINGS_ITEM["headerFlag"]] or int(
@ -448,7 +450,7 @@ def import_data_with_bulk(
LOAD DATA LOCAL LOAD DATA LOCAL
INFILE %s INFILE %s
REPLACE IGNORE
INTO TABLE {load_schema_name} INTO TABLE {load_schema_name}
CHARACTER SET {MYSQL_CHARSET_CODE[settings_list[SETTINGS_ITEM["charCode"]]]} CHARACTER SET {MYSQL_CHARSET_CODE[settings_list[SETTINGS_ITEM["charCode"]]]}
@ -482,24 +484,33 @@ def import_data_with_bulk(
# ロードスキーマのトランザクション開始 # ロードスキーマのトランザクション開始
with conn.cursor() as cur: with conn.cursor() as cur:
cur.execute(sql, [LOCAL_TEMPORARY_FILE_PATH, target_file_name]) cur.execute(sql, [LOCAL_TEMPORARY_FILE_PATH, target_file_name])
# 一括登録モードの場合、LOAD文の成功行数を取得してprocess_countにする # 処理件数と成功件数を取得する
cur.execute("SELECT ROW_COUNT()") cur.execute("SELECT ROW_COUNT()")
process_count = cur.fetchone()[0] normal_count = cur.fetchone()[0]
cur.execute("SELECT @file_row_cnt")
file_row_cnt = cur.fetchone()[0]
# ファイル行番号はヘッダがある時は-1して、データ行の行数と合わせる。
process_count = file_row_cnt - 1 if has_header else file_row_cnt
conn.commit() conn.commit()
except Exception as e: except Exception as e:
print( print(
f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {ERROR} E-MAIN-02 - 一括登録モードのSQL実行に失敗しました。エラー内容： {e}') f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {ERROR} E-MAIN-02 - 一括登録モードのSQL実行に失敗しました。エラー内容： {e}')
error(bucket_name, target_data_source, target_file_name, log_info) error(bucket_name, target_data_source, target_file_name, log_info)
# 一括登録の場合、クエリ実行に成功したら、処理件数と成功件数は同じにする # 処理件数 - 登録件数をワーニング件数とする
normal_count = process_count warning_count = process_count - normal_count
# ワーニングが1件でもある場合、一部登録ができていないため、ワーニング情報を登録する
if warning_count > 0:
warning_info = f'一部レコードの登録に失敗しています。行が重複している可能性があります。 投入データ件数：{process_count}, 正常終了件数：{normal_count}'
return { return {
"counts": { "counts": {
"process": process_count, "process": process_count,
"normal": normal_count, "normal": normal_count,
"warning": 0 # 一括登録時はワーニングにならずエラーになるため、必ず0件 "warning": warning_count
}, },
"warning_info": '' # 一括登録時はワーニングにならずエラーになるため、空文字を返却 "warning_info": warning_info
} }