Erstellen und Bereitstellen einer UDF mit Lambda

Um eine eigene UDF zu erstellen, erstellen Sie eine neue Java-Klasse, indem Sie die UserDefinedFunctionHandler-Klasse erweitern. Der Quellcode für das UserDefinedFunctionHandler.java im SDK ist auf GitHub im Repository awslabs/aws-athena-query-federation/athena-federation-sdk verfügbar, zusammen mit Beispiel-UDF-Implementierungen, die Sie untersuchen und ändern können, um eine eigene UDF zu erstellen.

Die Schritte in diesem Abschnitt veranschaulichen das Schreiben und Erstellen einer benutzerdefinierten UDF-JAR-Datei mit Apache Maven über die Befehlszeile und eine Bereitstellung.

Führen Sie die folgenden Schritte aus, um mit Maven eine benutzerdefinierte UDF für Athena zu erstellen

Klonen des SDK und Vorbereitung der Entwicklungsumgebung

Bevor Sie beginnen, stellen Sie mithilfe von sicher, dass git auf Ihrem System installiert ist sudo yum install git -y.

Erstellen Ihres Maven-Projekts

Führen Sie den folgenden Befehl aus, um Ihr Maven-Projekt zu erstellen. Ersetzen Sie groupId durch die eindeutige ID Ihrer Organisation und my-athena-udf durch den Namen Ihrer Anwendung. Weitere Informationen finden Sie unter Wie erstelle ich mein erstes Maven-Projekt? in der Apache Maven Dokumentation.

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

Hinzufügen von Abhängigkeiten und Plugins zu Ihrem Maven-Projekt

Fügen Sie die folgenden Konfigurationen zu Ihrer Maven-pom.xml-Projektdatei hinzu. Ein Beispiel finden Sie in der Datei pom.xml in 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>

Schreiben von Java-Code für die UDFs

Erstellen Sie eine neue Klasse, indem Sie UserDefinedFunctionHandler.java erweitern. Schreiben Sie Ihre UDFs in der Klasse.

Im folgenden Beispiel werden zwei Java-Methoden für UDFs, compress() und decompress(), innerhalb der Klasse MyUserDefinedFunctions erstellt.

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

Erstellen der JAR-Datei

Führen Sie mvn clean install aus um Ihr Projekt zu erstellen. Nachdem es erfolgreich erstellt wurde, wird eine JAR-Datei im target-Ordner Ihres Projekts mit dem Namen artifactId-version.jar erstellt, wobei artifactId der Name ist, den Sie im Maven-Projekt angegeben haben, z. B. my-athena-udfs.

Bereitstellen des JAR zu AWS Lambda

Sie haben zwei Möglichkeiten, Ihren Code in Lambda bereitzustellen:

Option 1: Bereitstellen zu AWS Serverless Application Repository

Wenn Sie Ihre JAR-Datei zu AWS Serverless Application Repository bereitstellen, erstellen Sie eine AWS SAM-YAML-Vorlagendatei, die die Architektur Ihrer Anwendung darstellt. Anschließend geben Sie diese YAML-Datei und einen Amazon-S3-Bucket an, in den Artefakte für Ihre Anwendung hochgeladen und für verfügbar gemacht werden AWS Serverless Application Repository. Im folgenden Verfahren wird das Skript publish.sh verwendet, das sich im athena-query-federation/tools-Verzeichnis des Athena-Query-Federation-SDKs befindet, das Sie zuvor geklont haben.

Weitere Informationen und Anforderungen finden Sie unter Veröffentlichen von Anwendungen im AWS Serverless Application Repository-Entwicklerhandbuch, unter AWS SAM-Vorlagenkonzepte im AWS Serverless Application Model-Entwicklerhandbuch und in Veröffentlichen von Serverless-Anwendungen mit der AWS SAM-CLI.

Das folgende Beispiel veranschaulicht Parameter in einer YAML-Datei. Fügen Sie Ihrer YAML-Datei ähnliche Parameter hinzu und speichern Sie diese in Ihrem Projektverzeichnis. Ein vollständiges Beispiel finden Sie unter athena-udf.yaml in 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

Kopieren Sie das publish.sh-Skript in das Projektverzeichnis, in dem Sie Ihre YAML-Datei gespeichert haben, und führen Sie den folgenden Befehl aus:

./publish.sh MyS3Location MyYamlFile

Zum Beispiel, wenn Ihr Bucket-Speicherort s3://amzn-s3-demo-bucket/mysarapps/athenaudf ist und Ihre YAML-Datei unter my-athena-udfs.yaml gespeichert wurde:

./publish.sh amzn-s3-demo-bucket/mysarapps/athenaudf my-athena-udfs
            

Option 2: Direktes Erstellen einer Lambda-Funktion

Sie können eine Lambda-Funktion auch direkt über die Konsole oder erstellen AWS CLI. Das folgende Beispiel veranschaulicht die Verwendung des Lambda-CLI-Befehls create-function.

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