Realtime Compute for Apache Flink:効率的なユーザー定義関数の記述

主要な実装ポイント

• 入力型と出力型として内部データ型を使用する。

  • 型変換を避けるために、Java Map の代わりにデータ入力に MapData を使用します。

  • データ出力に Java List の代わりに ArrayData を返し、メモリのオーバーヘッドをさらに削減します。

  public ArrayData eval(MapData input) {
      if (input == null) {
          return null;
      }
      return input.keyArray();
  }

• 入力型戦略：

  • 入力パラメーターが内部データ型に変換されていることを確認します。

  • 関数が 1 つのパラメーターのみを受け入れるように制限します。

  private static final InputTypeStrategy MAP = new InputTypeStrategy() {
      @Override
      public ArgumentCount getArgumentCount() {
          // 関数が 1 つのパラメーターのみを受け入れるように制限します。
          return ConstantArgumentCount.of(1);
      }
  
      @Override
      public Optional<List<DataType>> inferInputTypes(CallContext callContext, boolean throwOnFailure) {
          // 入力パラメーターが内部データ型に変換されていることを確認します。
          return Optional.of(
                  callContext.getArgumentDataTypes().stream()
                          .map(DataTypeUtils::toInternalDataType)
                          .collect(Collectors.toList()));
      }
  
      @Override
      public List<Signature> getExpectedSignatures(FunctionDefinition definition) {
          // シグネチャは返されません。通常は内部的に表示されます
          return null;
      }
  };

• 出力型戦略：

  • 出力型を配列として明示的に指定し、その要素型を入力 Map のキー型と一致させます。

  • 内部データ型を使用します。

      private static final TypeStrategy MAP_KEYS = callContext ->
          Optional.of(
              DataTypeUtils.toInternalDataType(DataTypes.ARRAY(
                  ((KeyValueDataType) callContext.getArgumentDataTypes().get(0)).getKeyDataType()
              ))
          );

使用上の注意

内部データ型はパフォーマンスブースターですが、UDF を記述する際には次の点を考慮してください。

• 操作の制限：内部データ型は、Java 型よりもサポートされる操作が少ないです。

• 急な学習曲線：Flink 内部データ型 API に関する知識が必要です。

• コードの可読性：標準の Java 型を使用する場合に比べて、コードが直感的ではありません。

• デバッグの難しさ：内部データ型のデバッグはより複雑になる可能性があります。