[airflow] Mysql to gcs to bigquery 할 때 CSV로 저장하기

이 글은 이전에 소개한 mysql => gcs, gcs => bigquery 를 이해한 상태에서 보는 것이 도움된다.

 

- [airflow] Mysql 데이터를 GCS(Google Cloud Storage)로 저장하기

lemontia.tistory.com/957

 

- [airflow] GCS(Google Cloud Storage) 파일을 BigQuery에 저장하기

lemontia.tistory.com/965

 

 

일전에 MySqlToGoogleCloudStorageOperator를 소개하면서 Json형태로 저장하는걸 소개했었는데, json은 포멧구조 상 key를 포함해야 하기 때문에 CSV에 비해 용량이 2배이상 차이난다.

 

json 형태

{"AGE": 25, "ID": 1, "NAME": "아이유", "REG_DT": 1598304038.0}
{"AGE": 33, "ID": 2, "NAME": "김수현", "REG_DT": 1598304338.0}
{"AGE": 18, "ID": 3, "NAME": "이수영", "REG_DT": 1598304638.0}

csv 형태

AGE,ID,NAME,REG_DT
25,1,"아이유",1598304038
33,2,"김수현",1598304338
18,3,"이수영",1598304638

 

그래서 CSV로 변환하여 저장하는 것을 추천한다.

 

설정방법은 다음과 같다.

mysqlToGCS = MySqlToGoogleCloudStorageOperator(
	task_id='mysql_to_gcs'
	, dag=dag
	, sql=[query 문]
	, bucket=[bucket 이름]
	, filename=[파일명]
	, mysql_conn_id=[접속정보(환경설정)]
	, export_format="csv"
	)

 

 

export_format 을 csv로 지정해야 한다.

 

Bigquery로 저장하는건 다음과 같이 설정하면 된다.

source_objects 를 설정할땐 앞뒤로 [] 가 있어야 한다

 

...
gcsToBigQuery = GoogleCloudStorageToBigQueryOperator(
		task_id='gcs_to_bq'
		, destination_project_dataset_table=[bigquery 테이블 명]
		, bucket=[bucket 이름]
		, source_objects=[[파일명]]
		, write_disposition='WRITE_TRUNCATE'
		, create_disposition='CREATE_IF_NEEDED'
		, source_format='CSV'
		, skip_leading_rows=1
		, dag=dag
		, schema_fields=[
			{'name': 'NAME', 'type': 'STRING', 'mode': 'REQUIRED'}
            ...
		]
	)
...

 

source_format 에 CSV를 넣고, skip_leading_rows 에는 1을 지정한다.

skip_leading_rows를 1로 지정한 이유는 첫번째 행이 제목이기 때문이다.

 

 

끝.