ãã®èšäºã¯ Timee Advent Calendar 2023 ã·ãªãŒãº 3 ã®12æ¥ç®ã®èšäºã§ãã qiita.com ã¯ããã« DREã°ã«ãŒãã§ããŒã¿ãšã³ãžãã¢ããã£ãŠãã 西山 ã§ãã ä»åã¯ãããŒã¿è»¢éãŸããã®éçšèªååã«ã€ããŠæžããŸãã ã¿ã€ããŒã®ã¢ããªãã°ãåæã§ããç¶æ
ã«ãªããŸã§ã®ãªãŒãã¿ã€ã ãé·ãã广æ€èšŒãæææ±ºå®ã«é
ããåºãŠããåé¡ã«å¯ŸããŠãdbtã«é¢é£ããéçšãèªååããããšã§æ¹åããŸããã ã¿ã€ããŒã§ã®ã¢ããªãã°ã®è»¢éã«ã€ã㊠ã¿ã€ããŒã§ã¯S3ã«è²¯ãŸã£ããµãŒããŒãã°ã宿çã«ããŒã¿åºç€(GCPã®BigQuery)ãžè»¢éããŠããããã°ãLakeå±€ãžè¿œå ããããšdbt(data build tool)ã«ãã£ãŠå倿çã®å å·¥åŠçããªãããŠStagingå±€ãžãšè»¢éãããŸãã ã¢ããªã®æ©èœè¿œå ã«ãã£ãŠæ°ãããã°ããŒãã«ã远å ãããå Žåãdbtã®åŠç远å ãå¿
èŠã«ãªããããGitHub Actionsã§å®æçã«Lakeå±€ãšStagingå±€ã§ãã°ããŒãã«ã®å·®ç°ããã§ãã¯ããLakeå±€ã«æ°ãã远å ãããããŒãã«ãããã°åŠç远å ãä¿ãéç¥ãslackãžéä¿¡ããŠããŸããã éç¥ã確èªãããã以äžã®å¯Ÿå¿ãè¡ããŸãã æ°ãã远å ãããããŒãã«ã®ã¹ããŒããç¢ºèª dbtã§modelãäœæ ä»ã®ãã°ããŒãã«ã®modelãã³ã㌠ã«ã©ã åããã£ã¹ãåŠçãä¿®æ£ {{ config( alias={{ ããŒãã«åãæžã }} ) }} {% set column_lists = [ {{ ã«ã©ã åãæžã }}] %} {{ production_logs_template({{ ããŒãã«åãæžã }}, column_lists) }} , result_record AS ( SELECT {{ ã«ã©ã åãæžã }}, dre_transfer_at, ROW_NUMBER() OVER (PARTITION BY {{ ã«ã©ã åãæžã }}) AS row_number FROM records ), stg_record AS ( SELECT {{ åã«ã©ã ã®å€æåŠçãæžã }}, FROM result_record WHERE row_number = 1 ) SELECT *, {{ set_record_exported_at(column_lists) }} AS dre__record_exported_at, FROM stg_record ããŒã å
ã¬ãã¥ãŒ dbtã®ãžã§ããå®è¡ããŠStgingå±€ã«ãããŒãã«ã远å åºæ¬çã«ã¯éç¥ãç¢ºèªæ¬¡ç¬¬ãäžèšã®å¯Ÿå¿ãè¡ãæ¹éã§ããããä»ã«åªå
床ã®é«ãé害察å¿ãä»ããŒã ããã®äŸé ŒããããšåŸåãã«ãªã£ãŠããŸãããšãå€ãã1é±é以å
ã«Stagingå±€ãžæ°èŠãã°ããŒãã«ã転éããåãæ±ºããšãªã£ãŠããŸããã ããããããã ãšãã°ãåæã§ããããã«ãªããŸã§ã®ãªãŒãã¿ã€ã ãé·ãããããã¯ãéçºããŒã ã®å¹ææ€èšŒãæææ±ºå®ã«é
ããåºãŠããŸããŸãã ããã§ãä»åã¯ãã®ãªãŒãã¿ã€ã ã®ççž®ãç®æãããšã«ããŸãã å®è£
æ¡ ã¢ããªã®ãã°ã«é¢ããŠã¯ãdbtã§å®è£
ããå å·¥åŠçã®å
容ãã»ãŒæ±ºãŸããã£ãŠãããããmodelã®è¿œå ãèªååã§ãããã§ãã (äŸ) æ«å°Ÿã«idãä»ãã«ã©ã ã¯INTåãžãã£ã¹ã æ«å°Ÿã«atãä»ãã«ã©ã ã¯JSTã®DATETIMEåãžãã£ã¹ã ãããŸã§æ°ãããã°ããŒãã«ãæ€ç¥ããŠããGitHub Actionsã®ããŒã«å
ã§ãdbtã®model远å ã®PRãçæããããšã«ããŸãã å®è£
ããŠã¿ã 倧ãŸãã«ãã£ãããšãšããŠã¯ä»¥äžã®éãã§ãã GitHub Apps ã«ããAPIãªã¯ãšã¹ãã®æºå pythonã§æžãããŠããç£èŠããŒã«ã®ä¿®æ£ dbtã®model(sqlãã¡ã€ã«)ãštest(ymlãã¡ã€ã«)ã®çæåŠç远å GitHubãžã®ãªã¯ãšã¹ãåŠç远å slackãžã®éç¥å
å®¹ä¿®æ£ GitHub Appsã«ããAPIãªã¯ãšã¹ãã®æºå ä»åã¯ãã°ããŒãã«ã®ç£èŠããŒã«ãšdbtãããžã§ã¯ããæ ŒçŽãããŠãããªããžããªãç°ãªã£ããããGitHub AppsãäœæããŠAPIæäœãã§ããããã«ããŸãã 以äžã®æé ã«æ²¿ã£ãŠAppãäœæ https://docs.github.com/ja/apps/creating-github-apps/registering-a-github-app/registering-a-github-app Appã®ç§å¯éµãååŸ https://docs.github.com/ja/apps/creating-github-apps/authenticating-with-a-github-app/managing-private-keys-for-github-apps èªèšŒã«å¿
èŠãª INSTALLATION_ID ãçæ https://docs.github.com/ja/apps/creating-github-apps/authenticating-with-a-github-app/authenticating-as-a-github-app-installation#generating-an-installation-access-token 3.ã§çæãã INSTALLATION_ID ã䜿ã£ãŠinstallation access tokenãçºè¡ ä»åã¯ç£èŠããŒã«ã«ä»¥äžã®ã¯ã©ã¹ã远å ããŸããã import json import os from datetime import datetime, timedelta import jwt import requests class GitHubAppTokenManager : def __init__ (self) -> None : self.GITHUB_APP_ID = os.environ.get( "APP_ID" ) self.GITHUB_APP_PRIVATE_KEY = os.environ.get( "APP_PRIVATE_KEY" ) self.GITHUB_APP_INSTALLATION_ID = os.environ.get( "APP_INSTALLATION_ID" ) def _generate_jwt (self) -> str : PEM = ( self.GITHUB_APP_PRIVATE_KEY.replace( " \\ n" , " \n " ) if self.GITHUB_APP_PRIVATE_KEY is not None else None ) utcnow = datetime.utcnow() alg = "RS256" payload = { "typ" : "JWT" , "alg" : alg, "iat" : utcnow, "exp" : utcnow + timedelta(seconds= 60 ), "iss" : self.GITHUB_APP_ID, } return jwt.encode(payload, PEM, algorithm=alg) def _generate_jwt_headers (self) -> dict : jwt_token = self._generate_jwt() return { "Authorization" : f "Bearer {jwt_token}" , "Accept" : "application/vnd.github.machine-man-preview+json" , } def _fetch_access_token (self) -> str : url = f "https://api.github.com/app/installations/{self.GITHUB_APP_INSTALLATION_ID}/access_tokens" response = requests.post(url, headers=self._generate_jwt_headers()) response.raise_for_status() return json.loads(response.text).get( "token" ) def generate_token_header (self) -> dict : token = self._fetch_access_token() return { "Authorization" : f "token {token}" , "Accept" : "application/vnd.github.inertia-preview+json" , } ãããŸã§ã§ããã°APIã®ãªã¯ãšã¹ãæºåå®äºã§ãã dbtã®modelãštestã®çæåŠç远å GitHub APIããªã¯ãšã¹ãããåã«ã³ããã察象ãšãªããã¡ã€ã«çæã®åŠçã远å ããŸãã ãŸãã¯modelã®sqlãã¡ã€ã«ãštestã®ymlãã¡ã€ã«ã®ãã³ãã¬ãŒããçšæããŸãã âmodelã®ãã³ãã¬ãŒãã§ãã dbtã§äœ¿ã£ãŠããJinjaã«åå¿ããŠæå³ãããšããã§å€ã眮ãæãããªããªãã®ã§ãäžéšãšã¹ã±ãŒãããŠããŸãã ãã£ã±ãå€å°èŠã¥ãããªãã®ã§ä»ã«è¯ãæ¹æ³ã¯ãªãã ãããã»ã»ã» {{ ' {{ ' }} config( alias= ' {{ table_name }} ' , ) {{ ' }} ' }} {{ ' {% ' }} set column_lists = {{ column_list }} {{ ' %} ' }} {{ ' {{ ' }} production_logs_template( ' {{ table_name }} ' , column_lists) {{ ' }} ' }} , result_record AS ( SELECT {{ select_list }}, dre_transfer_at, ROW_NUMBER() OVER (PARTITION BY {{ partition_column_list }}) AS row_number FROM records ), stg_record AS ( SELECT {{ jst_converted_select_list }}, dre_transfer_at, {{ ' {{ ' }} jst_now() {{ ' }} ' }} AS dre_delivered_at, FROM result_record WHERE row_number = 1 ) SELECT *, {{ ' {{ ' }} set_record_exported_at(column_lists) {{ ' }} ' }} AS dre__record_exported_at, FROM stg_record âtestã®ãã³ãã¬ãŒãã§ãããã°ã®äžææ§ãæ
ä¿ãããã¹ãã远å ããŸãã version : 2 models : - name : TABLE_NAME tests : - dbt_utils.unique_combination_of_columns : combination_of_columns : COLUMN_LISTS 次ã«ä»¥äžã®ã¯ã©ã¹ã远å ããŸãã BQãã远å 察象ããŒãã«ã®ã«ã©ã åãååŸã㊠generate_dbt_file_info 颿°ã«æž¡ãããšã§ãå
ã»ã©äœæãããã³ãã¬ãŒããå
ã«ãã¡ã€ã«ãçæãããŸãã import json import os import re import tempfile from typing import Any, Dict, List, Tuple from jinja2 import Environment, FileSystemLoader from ruamel.yaml import YAML class DBTModelFileGenerator : def _convert_column_with_at (self, column_name: str ) -> str : return f "TIMESTAMP({column_name})" def _convert_column_with_id (self, column_name: str ) -> str : return f "CAST({column_name} AS INT)" def _convert_utc_to_jst (self, column_name: str ) -> str : return f "{{{{ timestamp_utc2datetime_jst('{column_name}') }}}}" def _generate_select_list (self, column_list: List[ str ]) -> str : converted_columns = [] for col in column_list: if col.endswith( "_at" ): converted_columns.append( f "{self._convert_column_with_at(col)} as {col}" ) elif re.search( r"(_id$|^id$)" , col): converted_columns.append( f "{self._convert_column_with_id(col)} as {col}" ) else : converted_columns.append(col) return ", \n " .join(converted_columns) def _generate_jst_converted_select_list (self, column_list: List[ str ]) -> str : converted_list = [] for col in column_list: if col.endswith( "_at" ): converted_list.append(f "{self._convert_utc_to_jst(col)} as {col}" ) else : converted_list.append(col) return ", \n " .join(converted_list) def _generate_staging_sql (self, table_name: str , column_list: List[ str ]) -> str : select_list = self._generate_select_list(column_list) partition_column_list = ", " .join(column_list) jst_converted_select_list = self._generate_jst_converted_select_list( column_list ) # ãã³ãã¬ãŒãã®ããŒã file_loader = FileSystemLoader( "ãã³ãã¬ãŒããã¡ã€ã«ã®ãã¹" ) env = Environment(loader=file_loader) template = env.get_template( "template_stg_production_logs.sql" ) # ãã³ãã¬ãŒãã®ã¬ã³ããªã³ã° sql = template.render( table_name=table_name, column_list=column_list, select_list=select_list, partition_column_list=partition_column_list, jst_converted_select_list=jst_converted_select_list, ) # tempdirã«ãã¡ã€ã«ãä¿å output_path = os.path.join( tempfile.gettempdir(), f "{table_name}_stg_production_logs.sql" ) with open (output_path, "w" ) as tmp: tmp.write(sql) return output_path def _generate_staging_test (self, table_name: str , column_list: List[ str ]) -> str : yaml = YAML() yaml.indent(sequence= 4 , offset= 2 ) with open ( "ãã³ãã¬ãŒããã¡ã€ã«ã®ãã¹" , "r" , ) as template_file: template_content = yaml.load(template_file) template_content[ "models" ][ 0 ][ "name" ] = f "{table_name}_stg_production_logs" test_definition = template_content[ "models" ][ 0 ][ "tests" ][ 0 ] test_definition[ "dbt_utils.unique_combination_of_columns" ][ "combination_of_columns" ] = column_list output_path = os.path.join( tempfile.gettempdir(), f "{table_name}_stg_production_logs.yml" ) with open (output_path, "w" ) as tmp: yaml.dump(template_content, tmp) return output_path def generate_dbt_file_info ( self, new_table_and_column_names: List[Dict[ str , Any]] ) -> List[Tuple[ str , str , str ]]: files = [] for record in new_table_and_column_names: table_name = record[ "table_name" ] record_dict = json.loads(record[ "record" ]) column_list = list (record_dict.keys()) sql_file_path = self._generate_staging_sql(table_name, column_list) test_file_path = self._generate_staging_test(table_name, column_list) files.append( ( f "远å å
ã®dbt modelã®ãã¹/{table_name}_stg_production_logs.sql" , sql_file_path, f "Add staging SQL for {table_name}" , ) ) files.append( ( f "远å å
ã®dbt testã®ãã¹/{table_name}_stg_production_logs.yml" , test_file_path, f "Add staging test for {table_name}" , ) ) return files GitHubãžã®ãªã¯ãšã¹ãåŠç远å 以äžã®ã¯ã©ã¹ã远å ãã create_pr 颿°ã«å段ã§çæãããã¡ã€ã«ã®æ
å ±ãæž¡ãããšã§PRãçæãããŸãã èªèšŒã«ã¯APIã®ãªã¯ãšã¹ãæºåã§çºè¡ããinstallation access tokenã䜿ããŸãã import base64 import json import re from typing import Any, Dict, List, Tuple import jwt import requests from deleted_logs.adapter.output.github_app_token_manager import GitHubAppTokenManager class GitHubPRCreator : def __init__ (self, session_start_time: str ) -> None : self.OWNER = "XXX" self.REPO = "XXX" self.BASE_BRANCH_NAME = "main" numeric_only_session_start_time = re.sub( r"\D" , "" , session_start_time) self.NEW_BRANCH_NAME = f "feature/add_new_table_of_production_logs_to_staging_{numeric_only_session_start_time}" self.github_app_token_manager = GitHubAppTokenManager() def _is_branch_present (self) -> bool : headers = self.github_app_token_manager.generate_token_header() branch_url = f "https://api.github.com/repos/{self.OWNER}/{self.REPO}/git/ref/heads/{self.NEW_BRANCH_NAME}" response = requests.get(branch_url, headers=headers) if response.status_code == 200 : return True elif response.status_code == 404 : return False else : response.raise_for_status() return False def _create_branch (self) -> None : if self._is_branch_present(): return headers = self.github_app_token_manager.generate_token_header() # ããŒã¹ãã©ã³ãã®SHAãååŸ base_ref_url = f "https://api.github.com/repos/{self.OWNER}/{self.REPO}/git/ref/heads/{self.BASE_BRANCH_NAME}" response = requests.get(base_ref_url, headers=headers) response.raise_for_status() sha = response.json()[ "object" ][ "sha" ] # ãã©ã³ãäœæ branch_ref_url = ( f "https://api.github.com/repos/{self.OWNER}/{self.REPO}/git/refs" ) data = { "ref" : f "refs/heads/{self.NEW_BRANCH_NAME}" , "sha" : sha} response = requests.post(branch_ref_url, headers=headers, json=data) response.raise_for_status() def _upload_file_to_repository ( self, git_file_path: str , local_file_path: str , message: str ) -> None : headers = self.github_app_token_manager.generate_token_header() with open (local_file_path, "r" ) as file : content = file .read() encoded_content = base64.b64encode(content.encode()).decode() url = f "https://api.github.com/repos/{self.OWNER}/{self.REPO}/contents/{git_file_path}" # åæ§ã®ãã¡ã€ã«ãæ¢ã«ååšãããç¢ºèª response = requests.get(url, headers=headers) sha = None if response.status_code == 200 : sha = response.json()[ "sha" ] data = { "message" : message, "content" : encoded_content, "branch" : self.NEW_BRANCH_NAME, } if sha: data[ "sha" ] = sha response = requests.put(url, headers=headers, json=data) response.raise_for_status() def _push_files (self, files: List[Tuple[ str , str , str ]]) -> None : self._create_branch() for git_file_path, local_file_path, message in files: self._upload_file_to_repository(git_file_path, local_file_path, message) def _generate_pr_title (self, table_names: List[ str ]) -> str : return f "production_logsæ°èŠããŒãã«({','.join(table_names)})远å " def _generate_pr_body (self, table_and_column_names: List[Dict[ str , Any]]) -> str : header = "ãã®PRã¯deleted_logsã«ãã£ãŠèªåçæããããã®ã§ãã \n\n 以äžã®ããŒãã«ã®ã¹ããŒãžã³ã°åŠçã远å ããŠããŸãïŒ \n " lines = [] for record in table_and_column_names: table_name = record[ "table_name" ] record_dict = json.loads(record[ "record" ]) columns = ", " .join(record_dict.keys()) lines.append(f "**{table_name}** \n Columns: {columns} \n " ) return header + " \n " .join(lines) def create_pr ( self, table_names: List[ str ], column_names: List[Dict[ str , Any]], files: List[Tuple[ str , str , str ]], ) -> str : self._push_files(files) headers = self.github_app_token_manager.generate_token_header() url = f "https://api.github.com/repos/{self.OWNER}/{self.REPO}/pulls" title = self._generate_pr_title(table_names) body = self._generate_pr_body(column_names) data = { "title" : title, "body" : body, "head" : self.NEW_BRANCH_NAME, "base" : self.BASE_BRANCH_NAME, } response = requests.post(url, headers=headers, json=data) response.raise_for_status() return response.json()[ "html_url" ] ãããªæãã®PRãçæãããŸããã â»ãã¹ãã§äœæãããã®ãªã®ã§ã¯ããŒãºããŠãŸãã slackãžã®éç¥å
容ã®ä¿®æ£ åæ®µã§çæããPRã®URLãéç¥ã¡ãã»ãŒãžã«è¿œå ããŸãã âãã¹ãã§é£ã°ããéç¥ã¯ãããªæãã§ãã ããã§éç¥ãããPRã®ã¬ãã¥ãŒãšããŒãžãè¡ãããšã§ãStagingå±€ã®dbt modelãäœãããããã«ãªããŸããã ãŸãšã ä»ådbtã®modelçæãèªååããããšã§ã2ã€ã®å¹æãåŸãããŸããã 1ã€ç®ã¯ããã€ã«ã®åæžã§ãã ãããŸã§ä»¥äžã®ãããŒã§å¯Ÿå¿ããŠããŸãããã æ¢åã®dbt modelãã³ããããŠPRäœæ â ã¬ãã¥ãŒäŸé Œ â ããŒãž èªååããããšã§ã PRã確èªããŠããŒãž ã ãã§ã§ãããªããŸããã å
ã
倧ããå·¥æ°ã¯ããã£ãŠããªãã£ããã®ã®ãç¹°ãè¿ãçºçãã䟡å€ã®ãªãå®åäœæ¥ãæžããããšãã§ããŸããã ãããŠ2ã€ç®ã¯ãStagingå±€ãžæ°èŠãã°ããŒãã«ãåæ ããããŸã§ã®ã©ã°ççž®ã§ãã èªååããããšã§ãä»ã®å¯Ÿå¿ã§ã²ã£è¿«ããŠããéã«åŸåãã«ãããããšããªããªãã ãããŸã§ã¯ãã°åºåãå§ããŠããåæå¯èœã«ãªããŸã§æé·7æ¥ããã£ãŠãããšãããå°å
¥åŸã¯å€§äœ1æ¥ä»¥å
ã§åæå¯èœãªç¶æ
ã«ãªããŸããã å人çã«ã¯ãã¿ã€ããŒã«ãžã§ã€ã³ãããŸã§DevOpsçãªããšããããŠããªãã£ãããšããããéçšã¯å°å³ã§ã€ãŸããªãã€ã¡ãŒãžããã£ãã®ã§ãã(ãã¿ãŸãã)ããããã£ãŠæ¹åããŸããããšéçšãé¢çœããªãšæè¿æãããã«ãªããŸããã Weâre Hiring DREã°ã«ãŒãã§ã¯ãŸã ãŸã ãã£ãŠããããããšãããããããã®ã§ããããŸã ãŸã æãè¶³ããŠãããããšãã«åãã¡ã³ããŒãåéããŠããŸãïŒïŒ product-recruit.timee.co.jp