def users_transform(
    spark: SparkSession,
    dest_db_prefix: str,
    dest_table_name: str,
    execution_datetime_tag: Union[date, datetime],
) -> None:
    """
    1. Read "raw_users" and "raw_user_infos" tables
    2. preprocess "raw_users"
    3. preprocess "raw_user_infos"
    4. join on "user_id"
    5. write "users"
    """
    # some pyspark codes
    
    s3_path = os.path.join(
        S3_ROOT_PATH,
        f"{dest_db_prefix}_{db_suffix}",
        f"{dest_table_name}",
        f"{execution_datetime_tag.strftime('%Y_%m_%d_%H_%M_%S')}",
    )
    df.repartition(partition_num).write.mode("overwrite").option(
        "path", s3_path
    ).saveAsTable(f"{dest_db_prefix}_{db_suffix}.{dest_table_name}")

@click.command(name="users_transform")
@click.option("--dest_db_prefix", type=click.STRING, required=True)
@click.option("--dest_table_name", type=click.STRING, required=True)
@click.option("--airflow_execution_datetime_tag",type=click_utils.DateTimeParamType(),required=True,help="Airflow Execution Datetime",)
def cli_users_transform(dest_db_prefix: str,dest_table_name: str,airflow_execution_datetime_tag: Union[date, datetime],) -> None:
    """ 
    Join raw lectures and raw steps table, transform, write lectures
    """
    with spark_session(app_name="users_transform") as spark:
        users_transform(spark=spark,dest_db_prefix=dest_db_prefix,dest_table_name=dest_table_name,execution_datetime_tag=airflow_execution_datetime_tag,)

from cli_etl_codes import cli_users_transform

@click.group()
def run_task():
    """Simple group task that will group all sub-tasks together."""
    pass

# pyspark jobs
run_task.add_command(cli_users_transform)

dag = DAG(
    "transform-dag",
    schedule_interval="30 15 * * *",  # every 00:30 AM KST
    catchup=False,
    max_active_runs=1,
    default_args={
        **PYSPARK_ETL_DEFAULT_ARGS,
        "start_date": datetime(2020, 8, 6),
    },
)

users_transform = EMRSSHOperator(
    dag=dag,
    task_id="users_transform",
    image="my_imgage/pyspark-etl:latest",
    spark_program_args=shell_command_args(
        "users_transform",
        dest_db_prefix="etl",
        dest_table_name="users",
        execution_datetime_tag="",
    ),
)

class EMRSSHOperator(ssh_operator.SSHOperator):
    """
    Operator that inherits SSHOperator to run spark-submit command on remote EMR cluster
    
    """
    
    @decorators.apply_defaults
    def __init__(
        self,
        spark_program_args: str,
        driver_cores: str = "1",
        driver_memory: str = "2G",
        executor_cores: str = "1",
        executor_memory: str = "3G",
        spark_confs: list = [],
        extra_spark_conf_overrides: typing.Optional[dict] = None,
        image: str = None,
        *args: typing.Any,
        **kwargs: typing.Any,
    ) -> None:
        self.driver_cores = driver_cores
        self.driver_memory = driver_memory
        self.executor_cores = executor_cores
        self.executor_memory = executor_memory
        self.image = image
        self.spark_program_args = spark_program_args
        super().__init__(
            ssh_conn_id=f"my_emr_conn_ssh", command="", *args, **kwargs,
        )
        
    def execute(self, context: typing.Dict[str, typing.Any]) -> None:
        self.command = f"spark-submit \\
                --master yarn \\
                --deploy-mode cluster \\
                --conf spark.executorEnv.YARN_CONTAINER_RUNTIME_DOCKER_IMAGE={self.image} \\
                --conf spark.yarn.appMasterEnv.YARN_CONTAINER_RUNTIME_DOCKER_IMAGE={self.image} \\
                --executor-memory {self.executor_memory} \\
                --executor-cores {self.executor_cores} \\
                --driver-memory {self.driver_memory} \\
                --driver-cores {self.driver_cores} \\
                {self.spark_confs} \\
                run_task.py \\
                {self.spark_program_args}"
        super().execute(context)
        self.cleanup()

s3_path = os.path.join(
    "s3://my-data-lake",
    f"{dest_db_prefix}_{db_suffix}",
    f"{dest_table_name}",
    f"{execution_datetime_tag.strftime('%Y_%m_%d_%H_%M_%S')}",
)

df.repartition(partition_num).write.mode("overwrite").option(
    "path", s3_path
).saveAsTable(f"{dest_db_prefix}_{db_suffix}.{dest_table_name}")