SnowConvert AI - Teradata - TPT¶

このセクションでは、TeradataからSnowflakeへの TPT 翻訳について説明します。

TPT ステートメント変換¶

他のTeradataスクリプト言語と同様に、すべての TPT ステートメントはPythonコードに変換されます。以下はすでにサポートされている変換の例です。

ジョブヘッダー変換の定義¶

ジョブステートメントは、演算子、スキーマ定義、その内部のステップのようなすべてのステートメントを含むPythonクラスに翻訳されます。

ソースコード

 /* Some comments on the job  */
DEFINE JOB LOADJOB
DESCRIPTION 'LOAD AC_SCHEMA TABLE FROM A FILE'
JobBody

翻訳済みコード

 # Some comments on the job
class LOADJOB:
    # DESCRIPTION 'LOAD AC_SCHEMA TABLE FROM A FILE'
    JobBody

スキーマ変換の定義¶

スキーマステートメントは、ジョブステートメント用に作成されたクラスの属性に翻訳されます。

ソースコード

 DEFINE SCHEMA DCS_SCHEMA
DESCRIPTION 'DCS DATA'
(
PNRHEADER_ID   PERIOD(DATE),
PNRLOCPERIOD   PERIOD(TIMESTAMP(0)),
CRTDATE        CLOB,
REQTYP         JSON(100000),
seqno          INTEGER,
resdata        INTEGER
);

翻訳済みコード

 class JOBNAME:
    DCS_SCHEMA = """(
    PNRHEADER_ID VARCHAR(24),
    PNRLOCPERIOD VARCHAR(58),
    CRTDATE VARCHAR /*** MSC-WARNING - SSC-FDM-TD0002 - COLUMN CONVERTED FROM CLOB DATA TYPE ***/,
    REQTYP VARIANT,
    seqno INTEGER,
    resdata INTEGER,
    );"""

演算子変換の定義¶

演算子は、そのジョブのために生成されたクラス内で、python関数に変換されます。提供されている例は、 SnowConvert AI が現在サポートしている演算子です。

DDL 演算子¶

DDL 演算子のソースコード

 DEFINE OPERATOR DDL_OPERATOR()
DESCRIPTION 'TERADATA PARALLEL TRANSPORTER DDL OPERATOR'
TYPE DDL
ATTRIBUTES
(
  VARCHAR PrivateLogName ,
  VARCHAR TdpId          = @MyTdpId,
  VARCHAR UserName       = @MyUserName,
  VARCHAR UserPassword   = 'SomePassWord',
  VARCHAR AccountID,
  VARCHAR ErrorList      = ['3807','2580']
);

翻訳済みコード

 class JobName:
    def DDL_OPERATOR(self):
        #'TERADATA PARALLEL TRANSPORTER DDL OPERATOR'
        global args
        self.con = log_on(user = args.MyUserName, password = 'SomePassWord')

UPDATE 演算子¶

UPDATE 演算子のソースコード

 DEFINE OPERATOR LOAD_OPERATOR()
DESCRIPTION 'TERADATA PARALLEL TRANSPORTER LOAD OPERATOR'
TYPE UPDATE
SCHEMA AC_MASTER_SCHEMA
ATTRIBUTES
(
    VARCHAR PrivateLogName ,
    INTEGER MaxSessions       =  32,
    INTEGER MinSessions       =  1,
    VARCHAR TargetTable       = '&TARGET_TABLE',
    VARCHAR TdpId             = @MyTdpId,
    VARCHAR UserName          = @MyUserName,
    VARCHAR UserPassword      = @MyPassword,
    VARCHAR AccountId,
    VARCHAR ErrorTable1       = '&LOG_DB_NAME.ERR1',
    VARCHAR ErrorTable2       = '&LOG_DB_NAME.ERR2',
    VARCHAR LogTable          = '&LOG_DB_NAME.LOG_TABLE'
);

翻訳済みコード

 class JobName:
    def LOAD_OPERATOR(self, query):
        #'TERADATA PARALLEL TRANSPORTER LOAD OPERATOR'
        #USES SCHEMA AC_MASTER_SCHEMA
        operator_name = "LOAD_OPERATOR"
        return query

DATA CONNECTOR PRODUCER 演算子¶

Data Connector Producer演算子のソースコード

 DEFINE OPERATOR FILE_READER()
DESCRIPTION 'TERADATA PARALLEL TRANSPORTER DATA CONNECTOR OPERATOR'
TYPE DATACONNECTOR PRODUCER
SCHEMA AC_MASTER_SCHEMA
ATTRIBUTES
(
  VARCHAR PrivateLogName ,
  VARCHAR DirectoryPath   = '&INPUTFILEPATH' ,
  VARCHAR FileName        = '&INPUTTEXTFILE' ,
  VARCHAR Format          = 'delimited',
  VARCHAR OpenMode        = 'Read',
  VARCHAR TextDelimiter     = '~',
  VARCHAR IndicatorMode   = 'N'
);

翻訳済みコード

 class JobName:
    def FILE_READER(self):
        #'TERADATA PARALLEL TRANSPORTER DATA CONNECTOR OPERATOR'
        #USES SCHEMA AC_MASTER_SCHEMA
        operator_name = "FILE_READER"
        stage_name = f"{self.jobname}_{operator_name}"
        format_name = f"{self.jobname}_{operator_name}_FILEFORMAT"
        exec(f"""CREATE OR REPLACE FILE FORMAT {format_name} TYPE = 'CSV' FIELD_DELIMITER = '~' TRIM_SPACE = TRUE SKIP_HEADER = 0""")
        exec(f"""CREATE STAGE IF NOT EXISTS {self.jobname}_STAGE""")
        exec(f"""PUT file://{INPUTFILEPATH}/{INPUTTEXTFILE} @{stage_name} OVERWRITE = TRUE AUTO_COMPRESS = FALSE;""")
        temp_table_name = f"{self.jobname}_{operator_name}_TEMP"
        exec(f"""DROP TABLE IF EXISTS {temp_table_name}""")
        exec(f"""CREATE TEMPORARY TABLE {temp_table_name} {self.AC_MASTER_SCHEMA}""")
        exec(f"""COPY INTO {temp_table_name} FROM @{stage_name} FILE_FORMAT = (format_name = '{format_name}')""")
        return temp_table_name

ステップ変換の定義¶

ステップもジョブ用に生成されたクラス内のPython関数に翻訳され、翻訳されたコードのメイン関数で呼び出されます。

ステップのソースコード

 STEP setup_tables
(
  APPLY
  ('DELETE FROM  &STAGE_DB_NAME.EMS_AC_MASTER_STG;')
   TO OPERATOR (DDL_OPERATOR() );
);

STEP stLOAD_FILE_NAME
(
  APPLY
  ('INSERT INTO CRASHDUMPS.EMP_NAME
  (EMP_NAME, EMP_YEARS, EMP_TEAM)
  VALUES
  (:EMP_NAME, :EMP_YEARS, :EMP_TEAM);')
  TO OPERATOR (ol_EMP_NAME() [1])
  SELECT * FROM OPERATOR(op_EMP_NAME);
);

翻訳済みコード

 def setup_tables(self):
    self.DDL_OPERATOR()
    exec(f"""DELETE FROM DATABASE1.{STAGE_DB_NAME}.EMS_AC_MASTER_STG""")

def stLOAD_FILE_NAME(self):
    exec(f"""INSERT INTO DATABASE1.CRASHDUMPS.EMP_NAME (EMP_NAME, EMP_YEARS, EMP_TEAM)
SELECT EMP_NAME, EMP_YEARS, EMP_TEAM
FROM (
{self.ol_EMP_NAME('SELECT * FROM ' + self.op_EMP_NAME() )})""")

メイン関数¶

メイン関数はどのスクリプト言語でも常に生成されますが、 TPT の場合、メイン関数にはジョブクラスのインスタンスとジョブ内のステップの呼び出しが含まれます

メイン関数サンプルコード

 def main():
  _LOADJOB = LOADJOB()
  _LOADJOB.setup_tables()
  _LOADJOB.stLOAD_FILE_NAME()
  snowconvert.helpers.quit_application()