UDF使用 Lambda 建立和部署 - Amazon Athena
複製SDK並準備您的開發環境建立 Maven 專案將相依性和外掛程式新增至 Maven 專案撰寫以下項目的 Java 程式碼 UDFs建立檔JAR案部署JAR到 AWS Lambda

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

UDF使用 Lambda 建立和部署

若要建立自訂UDF，您可以藉由擴充類別來建立新的 Java UserDefinedFunctionHandler 類別。中 UserDefinedFunctionHandler.java 的原始程式碼可在 SDK awslabs/aws-athena-query-federation/athena-federation-sdk 儲存庫 GitHub 中取得，以及您可以檢查和修改以建立自訂的範例UDF實作。UDF

本節中的步驟演示了使用 Apache Maven 從命令行和部署編寫和構建自定義 UDF Jar 文件。

執行以下步驟，使用 Maven UDF 為 Athena 創建自定義

  1. 複製SDK並準備您的開發環境

  2. 建立 Maven 專案

  3. 將相依性和外掛程式新增至 Maven 專案

  4. 撰寫以下項目的 Java 程式碼 UDFs

  5. 建立檔JAR案

  6. 部署JAR到 AWS Lambda

複製SDK並準備您的開發環境

開始之前，務必使用 sudo yum install git -y 將 git 安裝在您的系統上。

若要安裝聯合 AWS 查詢 SDK

  • 在命令行中輸入以下內容以克隆存SDK儲庫。此儲存庫包括SDK、範例和一組資料來源連接器。如需資料來源連接器的詳細資訊，請參閱使用 Amazon Athena 聯合查詢。

    git clone https://github.com/awslabs/aws-athena-query-federation.git
安裝此程序的先決條件

如果您正在使用已安裝 Apache Maven AWS CLI、和 AWS Serverless Application Model 建置工具的開發機器，則可以略過此步驟。

  1. 從您複製時所建立的 aws-athena-query-federation 目錄的根目錄，執行 prepare_dev_env.sh 指令碼以準備開發環境。

  2. 更新 shell 以獲取安裝程序所建立的新變數，或重新啟動終端機工作階段。

    source ~/.profile
    重要

    如果跳過此步驟，稍後會收到有關 AWS CLI 或 AWS SAM 建置工具無法發佈 Lambda 函數的錯誤訊息。

建立 Maven 專案

執行以下命令來建立 Maven 專案。Replace (取代) groupId 使用您組織的唯一 ID，並替換 my-athena-udf 與您的應用程序的名稱有關更多信息，請參閱如何創建我的第一個 Maven 項目？ 在阿帕奇 Maven 的文檔。

mvn -B archetype:generate \
-DarchetypeGroupId=org.apache.maven.archetypes \
-DgroupId=groupId \
-DartifactId=my-athena-udfs

將相依性和外掛程式新增至 Maven 專案

將以下組態新增至 Maven 專案 pom.xml 檔檔案。如需範例，請參閱中的 pom.xml 檔案 GitHub。

<properties>
    <aws-athena-federation-sdk.version>2022.47.1</aws-athena-federation-sdk.version>
</properties>

<dependencies>
    <dependency>
        <groupId>com.amazonaws</groupId>
        <artifactId>aws-athena-federation-sdk</artifactId>
        <version>${aws-athena-federation-sdk.version}</version>
    </dependency>
</dependencies>
    
<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <version>3.2.1</version>
            <configuration>
                <createDependencyReducedPom>false</createDependencyReducedPom>
                <filters>
                    <filter>
                        <artifact>*:*</artifact>
                        <excludes>
                            <exclude>META-INF/*.SF</exclude>
                            <exclude>META-INF/*.DSA</exclude>
                            <exclude>META-INF/*.RSA</exclude>
                        </excludes>
                    </filter>
                </filters>
            </configuration>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>shade</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

撰寫以下項目的 Java 程式碼 UDFs

通過擴展 UserDefinedFunctionHandler.java 創建一個新的類。寫你的課堂UDFs裡面。

在下列範例中，會在類別內建立UDFsdecompress()、compress()和的兩個 Java 方法MyUserDefinedFunctions。

*package *com.mycompany.athena.udfs;

public class MyUserDefinedFunctions
        extends UserDefinedFunctionHandler
{
    private static final String SOURCE_TYPE = "MyCompany";

    public MyUserDefinedFunctions()
    {
        super(SOURCE_TYPE);
    }

    /**
     * Compresses a valid UTF-8 String using the zlib compression library.
     * Encodes bytes with Base64 encoding scheme.
     *
     * @param input the String to be compressed
     * @return the compressed String
     */
    public String compress(String input)
    {
        byte[] inputBytes = input.getBytes(StandardCharsets.UTF_8);

        // create compressor
        Deflater compressor = new Deflater();
        compressor.setInput(inputBytes);
        compressor.finish();

        // compress bytes to output stream
        byte[] buffer = new byte[4096];
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(inputBytes.length);
        while (!compressor.finished()) {
            int bytes = compressor.deflate(buffer);
            byteArrayOutputStream.write(buffer, 0, bytes);
        }

        try {
            byteArrayOutputStream.close();
        }
        catch (IOException e) {
            throw new RuntimeException("Failed to close ByteArrayOutputStream", e);
        }

        // return encoded string
        byte[] compressedBytes = byteArrayOutputStream.toByteArray();
        return Base64.getEncoder().encodeToString(compressedBytes);
    }

    /**
     * Decompresses a valid String that has been compressed using the zlib compression library.
     * Decodes bytes with Base64 decoding scheme.
     *
     * @param input the String to be decompressed
     * @return the decompressed String
     */
    public String decompress(String input)
    {
        byte[] inputBytes = Base64.getDecoder().decode((input));

        // create decompressor
        Inflater decompressor = new Inflater();
        decompressor.setInput(inputBytes, 0, inputBytes.length);

        // decompress bytes to output stream
        byte[] buffer = new byte[4096];
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(inputBytes.length);
        try {
            while (!decompressor.finished()) {
                int bytes = decompressor.inflate(buffer);
                if (bytes == 0 && decompressor.needsInput()) {
                    throw new DataFormatException("Input is truncated");
                }
                byteArrayOutputStream.write(buffer, 0, bytes);
            }
        }
        catch (DataFormatException e) {
            throw new RuntimeException("Failed to decompress string", e);
        }

        try {
            byteArrayOutputStream.close();
        }
        catch (IOException e) {
            throw new RuntimeException("Failed to close ByteArrayOutputStream", e);
        }

        // return decoded string
        byte[] decompressedBytes = byteArrayOutputStream.toByteArray();
        return new String(decompressedBytes, StandardCharsets.UTF_8);
    }
}

建立檔JAR案

執行 mvn clean install 來建置專案。成功構建後，將在名為的項目的JAR文件target夾中創建一個文件artifactId-version.jar，其中 artifactId 是您在 Maven 專案中提供的名稱，例如my-athena-udfs。

部署JAR到 AWS Lambda

將程式碼部署到 Lambda 時有兩種選項：

  • 部署方式 AWS Serverless Application Repository (建議)

  • 從JAR檔案建立 Lambda 函數

選項 1：部署到 AWS Serverless Application Repository

將JAR檔案部署到時 AWS Serverless Application Repository，您會建立代表應用程式架構的 AWS SAM 範本YAML檔案。然後，您可以指定此YAML檔案和 Amazon S3 儲存貯體，您應用程式的成品會在其中上傳並提供給 AWS Serverless Application Repository. 下列程序使用位於您先前複製的 Athena 查詢聯盟athena-query-federation/tools目錄中SDK的 publish.sh 指令碼。

如需詳細資訊和需求，請參閱AWS Serverless Application Repository 開發人員指南中的發佈應用程式、AWS Serverless Application Model 開發人員指南中的AWS SAM 範本概念，以及使用 AWS SAM CLI.

下面的例子演示了一個YAML文件中的參數。將類似的參數添加到您的YAML文件中，並將其保存在項目目錄中。有關完整示例，請參閱中的雅典娜-udf.yaml。 GitHub 

Transform: 'AWS::Serverless-2016-10-31'
Metadata:
  'AWS::ServerlessRepo::Application':
    Name: MyApplicationName
    Description: 'The description I write for my application'
    Author: 'Author Name'
    Labels:
      - athena-federation
    SemanticVersion: 1.0.0
Parameters:
  LambdaFunctionName:
    Description: 'The name of the Lambda function that will contain your UDFs.'
    Type: String
  LambdaTimeout:
    Description: 'Maximum Lambda invocation runtime in seconds. (min 1 - 900 max)'
    Default: 900
    Type: Number
  LambdaMemory:
    Description: 'Lambda memory in MB (min 128 - 3008 max).'
    Default: 3008
    Type: Number
Resources:
  ConnectorConfig:
    Type: 'AWS::Serverless::Function'
    Properties:
      FunctionName: !Ref LambdaFunctionName
      Handler: "full.path.to.your.handler. For example, com.amazonaws.athena.connectors.udfs.MyUDFHandler"
      CodeUri: "Relative path to your JAR file. For example, ./target/athena-udfs-1.0.jar"
      Description: "My description of the UDFs that this Lambda function enables."
      Runtime: java8
      Timeout: !Ref LambdaTimeout
      MemorySize: !Ref LambdaMemory

將指令publish.sh碼複製到儲存檔案的專YAML案目錄中，然後執行下列命令：

./publish.sh MyS3Location MyYamlFile

例如，如果您的值區位置是，s3://amzn-s3-demo-bucket/mysarapps/athenaudf且YAML檔案儲存為my-athena-udfs.yaml：

./publish.sh amzn-s3-demo-bucket/mysarapps/athenaudf my-athena-udfs
建立 Lambda 函式

  1. 開啟 Lambda 主控台 https://console.aws.amazon.com/lambda/，選擇 [建立函數]，然後選擇 [瀏覽無伺服器應用程式儲存庫]

  2. 選擇 Private applications (私有應用程式)，在清單中尋找您的應用程式，或使用關鍵字搜尋應用程式，然後選取應用程式。

  3. 檢閱並提供應用程式詳細資料，然後選擇 Deploy (部署)。

    您現在可以使用 Lambda 函數JAR檔案中定義的方法名稱，如 UDFs Athena 所示。

選項 2：直接建立 Lambda 函數

您也可以直接使用主控台或建立 Lambda 函數 AWS CLI。下列範例示範如何使用 Lambda create-function CLI 命令。

aws lambda create-function \
 --function-name MyLambdaFunctionName \
 --runtime java8 \
 --role arn:aws:iam::1234567890123:role/my_lambda_role \
 --handler com.mycompany.athena.udfs.MyUserDefinedFunctions \
 --timeout 900 \
 --zip-file fileb://./target/my-athena-udfs-1.0-SNAPSHOT.jar