マスタ管理システム課題管理

import sys

import re

import boto3

from decimal import *

from datetime import datetime, timedelta

from pyspark.sql import functions as F

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType

from awsglue.transforms import *

from awsglue.utils import getResolvedOptions

from pyspark.context import SparkContext

from awsglue.context import GlueContext

from awsglue.job import Job

import pytz

from pyspark.sql import DataFrame

import time

# Initialize Glue context and job

sc = SparkContext()

glueContext = GlueContext(sc)

spark = glueContext.spark_session

# Glue job initialization

job = Job(glueContext)

job.init("RecoveryJob", args={})

# Redshift connection options

redshift_connection_options = {

    "url": "jdbc:redshift://redshift.crlryjxenjmk.ap-northeast-1.redshift.amazonaws.com:5439/redshift_db",

    "user": "admin",

    "password": "Password123!",

    "truncate_table": "ON",

    "usestagingtable": "OFF",

    "redshiftTmpDir": "s3://data-platform-prd-glue-temp"

}

# Function to load a table from Redshift and create a temporary view

def load_redshift_table(dbtable, view_name):

    connection_options = redshift_connection_options.copy()

    connection_options["dbtable"] = dbtable

    df = glueContext.create_dynamic_frame_from_options("redshift", connection_options).toDF()

    df.createOrReplaceTempView(view_name)

    return df

# Load necessary tables from Redshift

load_redshift_table("dwh.dwh_forecast_tbl", "tbl_forecast")

load_redshift_table("dwh.dwh_jiseki_2019", "tbl_jiseki_2019")

load_redshift_table("dwh.dwh_mst_calendar", "tbl_calendar")

load_redshift_table("public.nec_tbl_401", "tbl_nec_401")

# Date range for processing

start_date = datetime.strptime('20241115', '%Y%m%d')

end_date = datetime.strptime('20241118', '%Y%m%d')

current_date = start_date

while current_date <= end_date:

    try:

        # Calculate previous year date

        ly_today = current_date.replace(year=current_date.year - 1)

        # Calculate 2019 date

        today_2019 = ly_today.replace(year=2019)

        # 修正部分: pre_year_cal_date_ymdを使用して前年の日付を取得

        pre_year_date_query = f"""

            SELECT pre_year_cal_date_ymd 

            FROM tbl_calendar 

            WHERE cal_date_ymd = '{current_date.strftime('%Y%m%d')}'

        """

        pre_year_date_df = spark.sql(pre_year_date_query)

        pre_year_date = datetime.strptime(pre_year_date_df.collect()[0][0], '%Y%m%d')

        # Create parameter set for current date

        params = {

            "today": current_date.strftime('%Y%m%d'),

            "ly_today": ly_today.strftime('%Y%m%d'),

            "2019_today": today_2019.strftime('%Y%m%d'),

            "pre_year_date": pre_year_date.strftime('%Y%m%d')

        }

        # Main SQL query with tbl_forecast as the base

        sql_nec = f"""

        SELECT

            fc.date AS date,

            fc.mst_shop_shop_cd AS shop_code,

            COALESCE(n1.n1_amount_counter, 0) AS actual_sales,

            COALESCE(n2.n2_customer_counter, 0) AS actual_customer_num,

            COALESCE(fc.sales_budget, 0) AS budget_sales,

            COALESCE(fc.customer_budget, 0) AS budget_customer_num,

            COALESCE(n4.n4_amount_counter, 0) AS last_year_sales,

            COALESCE(n5.n5_customer_counter, 0) AS last_year_customer_num,

            COALESCE(n6.n6_amount_counter, 0) AS last_year_sales_2,

            COALESCE(n7.n7_customer_counter, 0) AS last_year_customer_num_2,

            COALESCE(n8.sales_2019, 0) AS sales_2019,

            COALESCE(n8.customer_2019, 0) AS customer_2019_num,

            COALESCE(n9.sales_2019, 0) AS sales_2019_2,

            COALESCE(n9.customer_2019, 0) AS customer_2019_num_2

        FROM 

            tbl_forecast fc

        LEFT JOIN 

            (

                SELECT

                    tz.transaction_date,

                    tz.store_code,

                    SUM(tz.amount_counter) AS n1_amount_counter

                FROM tbl_nec_401 tz

                WHERE REPLACE(tz.transaction_date, '-', '') = '{params["today"]}'

                GROUP BY tz.transaction_date, tz.store_code

            ) n1 

        ON SUBSTRING(n1.store_code, 1, 6) = fc.mst_shop_shop_cd

        LEFT JOIN 

            (

                SELECT

                    tz.store_code,

                    SUM(tz.customer_counter) AS n2_customer_counter

                FROM tbl_nec_401 tz

                WHERE REPLACE(tz.transaction_date, '-', '') = '{params["today"]}'

                GROUP BY tz.store_code

            ) n2

        ON SUBSTRING(n2.store_code, 1, 6) = fc.mst_shop_shop_cd

        LEFT JOIN 

            (

                SELECT

                    tz.store_code,

                    SUM(tz.amount_counter) AS n4_amount_counter

                FROM tbl_nec_401 tz

                WHERE REPLACE(tz.transaction_date, '-', '') = '{params["ly_today"]}'

                GROUP BY tz.store_code

            ) n4

        ON SUBSTRING(n4.store_code, 1, 6) = fc.mst_shop_shop_cd

        LEFT JOIN 

            (

                SELECT

                    tz.store_code,

                    SUM(tz.customer_counter) AS n5_customer_counter

                FROM tbl_nec_401 tz

                WHERE REPLACE(tz.transaction_date, '-', '') = '{params["ly_today"]}'

                GROUP BY tz.store_code

            ) n5

        ON SUBSTRING(n5.store_code, 1, 6) = fc.mst_shop_shop_cd

        LEFT JOIN 

            (

                SELECT

                    tz.store_code,

                    SUM(tz.amount_counter) AS n6_amount_counter

                FROM tbl_nec_401 tz

                WHERE REPLACE(tz.transaction_date, '-', '') = '{params["pre_year_date"]}'

                GROUP BY tz.store_code

            ) n6

        ON SUBSTRING(n6.store_code, 1, 6) = fc.mst_shop_shop_cd

        LEFT JOIN 

            (

                SELECT

                    tz.store_code,

                    SUM(tz.customer_counter) AS n7_customer_counter

                FROM tbl_nec_401 tz

                WHERE REPLACE(tz.transaction_date, '-', '') = '{params["pre_year_date"]}'

                GROUP BY tz.store_code

            ) n7

        ON SUBSTRING(n7.store_code, 1, 6) = fc.mst_shop_shop_cd

        LEFT JOIN 

            (

                SELECT

                    js.store_code,

                    js.sales_2019,

                    js.customer_2019

                FROM tbl_jiseki_2019 js

                WHERE js.date = '{params["2019_today"]}'

            ) n8

        ON SUBSTRING(n8.store_code, 1, 6) = fc.mst_shop_shop_cd

        LEFT JOIN 

            (

                SELECT

                    js.store_code,

                    js.sales_2019,

                    js.customer_2019

                FROM tbl_jiseki_2019 js

                WHERE js.date = (

                    SELECT year_2019_cal_date_ymd 

                    FROM tbl_calendar 

                    WHERE cal_date_ymd = '{params["today"]}'

                )

            ) n9

        ON SUBSTRING(n9.store_code, 1, 6) = fc.mst_shop_shop_cd

        WHERE 

            fc.date = '{params["today"]}'

            AND fc.mst_shop_shop_cd > '001000'

        """

        # Execute the SQL query

        df_result_nec = spark.sql(sql_nec)

        # Write the result to MySQL

        def write_to_mysql(df: DataFrame):

            for attempt in range(5):

                try:

                    df.write.format("jdbc") \

                        .option(

                            "url", 

                            "jdbc:mysql://data-platform-prd-aurora.cluster-cy9zohy3hxcg.ap-northeast-1.rds.amazonaws.com:3306/dm"

                        ) \

                        .option("driver", "com.mysql.jdbc.Driver") \

                        .option("dbtable", "flash_sales_reports") \

                        .option("user", "root") \

                        .option("password", "Password123!") \

                        .option("characterEncoding", "UTF8") \

                        .mode("Append") \

                        .save()

                    print(f"Successfully wrote data for date {params['today']} to MySQL.")

                    break

                except Exception as e:

                    print(f"Attempt {attempt + 1} failed: {str(e)}")

                    if attempt < 4:

                        time.sleep(5)  # Wait before retrying

                    else:

                        raise

        # Write the result to MySQL

        write_to_mysql(df_result_nec)

    except Exception as e:

        print(f"Error processing date {current_date.strftime('%Y%m%d')}: {str(e)}")

    # Move to the next date

    current_date += timedelta(days=1)

# Commit the Glue job

job.commit()