SnowConvert AI – Teradata – TPT¶

Esta seção ilustra a tradução TPT do Teradata para Snowflake.

Transformação de instruções TPT¶

Todas as instruções TPT, assim como outras linguagens de script do Teradata, estão sendo convertidas em código python. Aqui estão alguns exemplos de transformações já suportadas.

Defina a transformação do cabeçalho de trabalho¶

A instrução de trabalho é convertida para uma classe python com todas as instruções, como operadores, definições de esquema e etapas dentro dela.

Código-fonte

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

Copy

Código convertido

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

Copy

Defina a transformação do esquema¶

A instrução de esquema é convertida em um atributo na classe criada para a instrução de trabalho.

Código-fonte

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

Copy

Código convertido

 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,
    );"""

Copy

Defina a transformação do operador¶

Os operadores são traduzidos para funções Python dentro da classe gerada para o trabalho. Os exemplos fornecidos são os operadores a que o SnowConvert AI atualmente oferece suporte

Operador DDL¶

Código-fonte do operador 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']
);

Copy

Código convertido

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

Copy

Operador UPDATE¶

Código-fonte do operador 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'
);

Copy

Código convertido

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

Copy

Operador DATA CONNECTOR PRODUCER¶

Código-fonte do operador 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'
);

Copy

Código convertido

 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

Copy

Defina transformação de etapas¶

As etapas também são convertidas para funções python dentro da classe gerada para o trabalho; elas serão chamadas na função principal do código convertido.

Código-fonte da etapa

 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);
);

Copy

Código convertido

 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() )})""")

Copy

Função principal¶

A função principal é sempre gerada para qualquer linguagem de script; para TPT, a função principal contém uma instância da classe de trabalho e chamadas para as etapas do trabalho

Código de amostra da função principal

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

Copy