ユーザー定義関数を実装する

ユーザー定義関数 (UDF) は、頻繁に使用されるロジックや、他の方法ではクエリで表現できないカスタムロジックを呼び出すことができる拡張ポイントです。Python または Java や Scala などの JVM 言語を使用して、Studio ノートブック内の段落に UDF を実装できます。JVM 言語で実装された UDF を含む外部 JAR ファイルを Studio ノートブックに追加することもできます。

UserDefinedFunction　( または独自の抽象クラス ） をサブクラス化する抽象クラスを登録する JAR を実装する場合は、Apache Maven で提供されている範囲、Gradle の compileOnly 依存宣言、SBT で提供されている範囲、または UDF プロジェクトのビルド設定で同等の命令を使用します。これにより、UDF ソースコードは Flink API に対してコンパイルできますが、Flink API クラス自体はビルドアーティファクトに含まれません。Maven プロジェクトでこのような前提条件を守っている UDF jar の例の「pom」を参照してください。

注記

セットアップの例については、「AWS Machine Learning Blog」の「Amazon Managed Service for Apache Flink、Amazon Translate および Amazon Comprehend を搭載した SQL 関数を使用してストリーミングデータを翻訳、修正、分析する」を参照してください。

コンソールを使用して UDF JAR ファイルを Studio ノートブックに追加するには、以下の手順に従います。

1. UDF JAR ファイルを Amazon S3 にアップロードします。

2. で AWS Management Console、Studio ノートブックを作成するためのカスタム作成オプションを選択します。

3. Studio ノートブック作成ワークフローに従って 設定 ステップまで進みます。

4. [ユーザー定義関数] セクションで、[ユーザー定義関数を追加] を選択します。

5. UDF が実装されている JAR ファイルまたは ZIP ファイルの Amazon S3 ロケーションを指定します。

6. [Save changes] (変更の保存) をクリックします。

CreateApplication API を使用して新しい Studio ノートブックを作成するときに UDF JAR を追加するには、 CustomArtifactConfiguration データタイプで JAR のロケーションを指定します。UDF JAR を既存の Studio ノートブックに追加するには、UpdateApplication API オペレーションを呼び出し、 CustomArtifactsConfigurationUpdate データタイプで JAR のロケーションを指定します。または、 AWS Management Console を使用して UDF JAR ファイルを Studio ノートブックに追加することもできます。

ユーザー定義関数に関する考慮事項

• Apache Flink Studio 用 Managed Service では、「Apache Zeppelin の用語」が使われています。ここで、ノートブックは、複数のノートを含むことができる Zeppelin インスタンスです。これにより、各ノートには複数の段落を含めることができます。Apache Flink Studio 用 Managed Service では、インタープリタープロセスはノートブックのすべてのノートで共有されます。そのため、あるノートで「CreateTemporarySystemFunction」を使用して明示的な関数登録を実行すると、同じノートブックの別のノートでも同じ関数をそのまま参照できます。

  ただし、「アプリケーションとしてデプロイ」操作は、ノートブック内のすべてのノートではなく、「個々」のノートに適用されます。アプリケーションとしてデプロイすると、アクティブノートの内容のみがアプリケーションの生成に使用されます。他のノートブックで行われる明示的な関数登録は、生成されるアプリケーションの依存関係の一部ではありません。さらに、アプリケーション・オプションとしてデプロイする際に、JAR のメインクラス名を小文字の文字列に変換することで、暗黙的な関数登録が行われます。

  たとえば、 TextAnalyticsUDF が UDF JAR のメインクラスである場合、暗黙的に登録すると関数名 textanalyticsudf になります。そのため、Studio のノート 1 に次のような明示的な関数登録があった場合、共有インタープリタがあるため、そのノート（例えばノート 2）の他のすべてのノートは、名前 myNewFuncNameForClass でその関数を参照することができ

  stenv.createTemporarySystemFunction("myNewFuncNameForClass", new TextAnalyticsUDF())

  しかし、ノート 2 のアプリケーションとしてデプロイする操作では、この明示的な登録は依存関係に「含まれない」ため、デプロイされたアプリケーションは期待通りに動作しません。暗黙的に登録されるため、デフォルトでは、この関数へのすべての参照は myNewFuncNameForClass ではなく textanalyticsudf であることが想定されます。

  カスタム関数名を登録する必要がある場合は、ノート 2 自体に別の段落を設け、次のように明示的に登録することが想定されます。

  %flink(parallelism=l)
  import com.amazonaws.kinesis.udf.textanalytics.TextAnalyticsUDF 
  # re-register the JAR for UDF with custom name
  stenv.createTemporarySystemFunction("myNewFuncNameForClass", new TextAnalyticsUDF())
  

  %flink. ssql(type=update, parallelism=1) 
  INSERT INTO
      table2
  SELECT
      myNewFuncNameForClass(column_name)
  FROM
      table1
  ;

• UDF JAR に Flink SDK が含まれている場合は、UDF ソースコードが Flink SDK に対してコンパイルできるように Java プロジェクトを設定しますが、Flink SDK クラス自体はビルドアーティファクト（JAR など）に含まれません。

  Apache Maven の provided スコープ、Gradle の compileOnly 依存宣言、SBT の provided スコープ、または UDF プロジェクトのビルド設定で同等の命令を使用できます。この「pom」は、Maven プロジェクトでこのような前提条件を満たす UDF jar の例から参照できます。詳細な手順のチュートリアルについては、「Amazon Managed Service for Apache Flink、Amazon Translate、Amazon Comprehend で SQL 関数を使用してストリーミングデータを翻訳、編集、分析する」を参照してください。