基于Lamda的宾得ETL变换

Question

是否可以使用AWS函数运行宾得ETL作业/转换？

我有五旬节ETL作业按计划在Windows服务器上运行，我们计划迁移到AWS。我在考虑Lambda函数。只是为了了解是否有可能使用AWS Lamdba来安排五旬节ETL作业

Answer 1

下面是我能够在AWS函数中成功运行的代码片段。

从AWS函数调用handleRequest函数

    public Integer handleRequest(String input, Context context) {
        parseInput(input);
        return executeKtr(transName);

    }

parseInput:此函数用于解析由Lambda函数传递的字符串参数，以提取KTR名称及其带有值的参数。输入格式为"ktrfilename param1=value1 param2=value2“。

    public static void parseInput(String input) {
        String[] tokens = input.split(" ");     
        transName = tokens[0].replace(".ktr", "") + ".ktr";
        for (int i=1; i<tokens.length; i++) {
            params.add(tokens[i]);
        }
    }

执行KTR:我使用git存储所有KTR文件，并根据作为参数KTR传递的名称执行

    public static Integer executeKtr(String ktrName) {
        try {
            System.out.println("Present Project Directory : " + System.getProperty("user.dir"));

            String transName = ktrName.replace(".ktr", "") + ".ktr";
            String gitURI = awsSSM.getParaValue("kattle-trans-git-url");
            String repoLocalPath = clonePDIrepo.cloneRepo(gitURI);

            String path = new File(repoLocalPath + "/" + transName).getAbsolutePath();
            File ktrFile = new File(path);

            System.out.println("KTR Path: " + path);
            try {
                /**
                 * IMPORTANT NOTE FOR LAMBDA FUNCTION MUST CREATE .KEETLE DIRECOTRY OTHERWISE
                 * CODE WILL FAIL IN LAMBDA FUNCTION WITH ERROR CANT CREATE
                 * .kettle/kettle.properties file.
                 * 
                 * ALSO SET ENVIRNOMENT VARIABLE ON LAMBDA FUNCTION TO POINT
                 * KETTLE_HOME=/tmp/.kettle
                 */
                Files.createDirectories(Paths.get("/tmp/.kettle"));
            } catch (IOException e) {
                e.printStackTrace();
                throw new RuntimeException("Error Creating /tmp/.kettle directory");

            }

            if (ktrFile.exists()) {
                KettleEnvironment.init();
                TransMeta metaData = new TransMeta(path);
                Trans trans = new Trans(metaData);
//           SETTING PARAMETERS
                trans = parameterSetting(trans);

                trans.execute( null );
                trans.waitUntilFinished();
                if (trans.getErrors() > 0) {
                    System.out.print("Error Executing transformation");
                    throw new RuntimeException("There are errors in running transformations");
                } else {
                    System.out.print("Successfully Executed Transformation");
                    return 1;
                }
            } else {
                System.out.print("KTR File:" + path + " not found in repo");
                throw new RuntimeException("KTR File:" + path + " not found in repo");
            }

        } catch (KettleException e) {
            e.printStackTrace();
            throw new RuntimeException(e.getMessage());
        }
    }

parameterSetting:如果KTR接受参数，并且在调用AWS函数时传递参数，则使用parameterSetting函数设置它。

    public static Trans parameterSetting(Trans trans) {
        String[] transParams = trans.listParameters();

        for (String param : transParams) {
            for (String p: params) {
                String name = p.split("=")[0];
                String val = p.split("=")[1];
                
                if (name.trim().equals(param.trim())) {
                    try {
                        System.out.println("Setting Parameter:"+ name + "=" + val);
                        trans.setParameterValue(name, val);
                    } catch (UnknownParamException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
        }
        trans.activateParameters();
        return trans;
    }

CloneGitRepo：

public class clonePDIrepo {
    /**
     * Clones the given repo to local folder
     *
     * @param pathWithPwd Gir repo URL with access token included in the url. e.g.
     *                    https://token_name:token_value@github.com/ktr-git-repo.git
     * @return returns Local Repository String Path
     */
    public static String cloneRepo(String pathWithPwd) {
        try {
            /**
             * CREATING TEMP DIR TO AVOID FOLDER EXISTS ERROR, THIS TEMP DIRECTORY LATER CAN
             * BE USED TO GET ABSOLETE PATH FOR FILES IN DIRECTORY
             */
            File pdiLocalPath = Files.createTempDirectory("repodir").toFile();
            Git git = Git.cloneRepository().setURI(pathWithPwd).setDirectory(pdiLocalPath).call();
            System.out.println("Git repository cloned successfully");
            System.out.println("Local Repository Path:" + pdiLocalPath.getAbsolutePath());

            // }
            return pdiLocalPath.getAbsolutePath();
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }
}

AWSSSMgetParaValue:获取传递的参数的字符串值。

    public static String getParaValue(String paraName) {

        try {
            Region region = Region.US_EAST_1;
            SsmClient ssmClient = SsmClient.builder()
                    .region(region)
                    .build();

            
            GetParameterRequest parameterRequest = GetParameterRequest.builder()
                .name(paraName)
                .withDecryption(true)
                .build();
            GetParameterResponse parameterResponse = ssmClient.getParameter(parameterRequest);
            
            System.out.println(paraName+ " value retreived from AWS SSM");
            ssmClient.close();
            return parameterResponse.parameter().value();

        } catch (SsmException e) {
        System.err.println(e.getMessage());
        return null;
        }
   }

假设：

file

• Environment是用KTR文件创建的，KETTLE_HOME=/tmp/.kettle

• Lambda函数的根上存在着git url，存在于aws上，具有有效的令牌来克隆回购Network

• Proper输入字符串，其中包含KTR的KTR变量的名称，因为KETTLE_HOME=/tmp/.kettle

• Lambda函数对SSM具有必要的权限，并设置了S3 VPC安全组规则，以允许对KTR文件

进行必要的网络访问。

我计划上传完整的代码到git。我将用存储库的URL更新这个帖子。