sean-azoci

#쿼리 문법 점검
client = get_bigquery_client()
job_config = bigquery.QueryJobConfig(dry_run=True, use_legacy_sql=False)
try:
	query_job = client.query(
			query=query,
			job_config=job_config)
except Exception as e:
	raise Exception(f'{file_name} SQL Syntax is not valid.')

sean-azoci

graph = defaultdict(list)
visited = defaultdict(int)

for node in list(graph):
  if not visited[node]:
    if not dfs(graph, visited, node):
      raise Exception(f'{node} has not acyclic dependencies.')

sean-azoci

#쿼리 문법 점검
client = get_bigquery_client()
job_config = bigquery.QueryJobConfig(dry_run=True, use_legacy_sql=False)
try:
  query_job = client.query(
    query=query,
    job_config=job_config)
except Exception as e:
  raise Exception(f'{file_name} SQL Syntax is not valid.')

sean-azoci

json_array = Variable.get(variable, deserialize_json=True)
filtered = [x for x in json_array if x['is_enabled'] == 'true']
for item in filtered:
  ecs_dict = get_ecs_operator_dict(service_name, item, date_kst)
  ecs_task = ECSOperator(
      **ecs_dict
  )

  gcs_del_dict = get_gcs_delete_prefix_dict(service_name, item, date_kst)
  gcs_del_task = GoogleCloudStorageDeleteOperator(

sean-azoci

{
  "question": [
    {
      "event_name": "question"
    }
  ],
  "question_payment": [
    {
      "event_name": "question_payment"
    }

sean-azoci

command = [
  "spark-2.4.5-bin-hadoop2.7/bin/spark-submit",
  "--class",
  get_class_name(item_json["etl_type"]),
  "--master",
  "local[*]",
  "--driver-memory",
  resource_type["driver"],
  "qanda-data-ecs-task-assembly-0.1.0-SNAPSHOT.jar",
  service_name,

sean-azoci

#디펜던시 코드
{
	'mart_a' : ['start', 'mart_d'],
	'mart_b' : ['start'],
	'mart_c' : ['mart_a', 'mart_b'],
	'mart_d' : ['mart_c', 'mart_b']
}


#순환을 찾는 코드 예

sean-azoci

client = get_bigquery_client()
job_config = bigquery.QueryJobConfig(dry_run=True, use_query_cache=False, use_legacy_sql=False)
try:
	query_job = client.query(
      query=query,
      job_config=job_config)
except Exception as e:
  raise Exception(f'{file_name} SQL Syntax is not valid.')

sean-azoci

#파일명
{dataset}/{table_name}.sql

#태스크명
{dataset}_{table_name}

#테이블명
{dataset}.{table_name}

sean-azoci

#Dag 내 실행 코드 예시
variable = '{service_name}'
tables = Variable.get(variable, deserialize_json=True)
for table in tables:
  spark_task = Spark_job_Operator(task_id = 'fspark_{table}' ...)
  load_task = Python_Operator(task_id = f'load_{table}' ...)
  spark_task >> load_task