翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

# Managed Service for Apache Flink の Python アプリケーションをプログラムする
<a name="how-python-programming"></a>

Python アプリケーション向けのApache Flink 用 Managed Serviceは、Apache Flink Python テーブル API を使用してコーディングします。Apache Flink エンジンは Python テーブル API ステートメント (Python VM で実行されている) を Java テーブル API ステートメント (Java VM で実行されている) に変換します。

Python テーブル API を使用するには、次の操作を行います。
+ `StreamTableEnvironment` へのリファレンスを作成します。
+ `StreamTableEnvironment` リファレンスに対してクエリを実行して、`table` ソースストリーミングデータからオブジェクトを作成します。
+ `table` オブジェクトに対してクエリを実行して出力テーブルを作成します。
+ `StatementSet` を使用して出力テーブルを宛先に書き込みます。

Apache Flink 用 Managed Serviceで Python テーブル API を使い始めるには、[Python 用 Amazon Managed Service for Apache Flink 入門](gs-python.md) を参照してください。

## ストリーミングデータの読み取りと書き込み
<a name="how-python-programming-readwrite"></a>

ストリーミングデータを読み書きするには、テーブル環境で SQL クエリを実行します。

### テーブルを作成する
<a name="how-python-programming-readwrite-createtable"></a>

次のコード例は、SQL クエリを作成するユーザー定義関数を示しています。SQL クエリは Kinesis ストリームと相互作用するテーブルを作成します。

```
def create_table(table_name, stream_name, region, stream_initpos):
   return """ CREATE TABLE {0} (
                `record_id` VARCHAR(64) NOT NULL,
                `event_time` BIGINT NOT NULL,
                `record_number` BIGINT NOT NULL,
                `num_retries` BIGINT NOT NULL,
                `verified` BOOLEAN NOT NULL
              )
              PARTITIONED BY (record_id)
              WITH (
                'connector' = 'kinesis',
                'stream' = '{1}',
                'aws.region' = '{2}',
                'scan.stream.initpos' = '{3}',
                'sink.partitioner-field-delimiter' = ';',
                'sink.producer.collection-max-count' = '100',
                'format' = 'json',
                'json.timestamp-format.standard' = 'ISO-8601'
              ) """.format(table_name, stream_name, region, stream_initpos)
```

### ストリーミングデータの読み取り
<a name="how-python-programming-readwrite-read"></a>

次のコード例は、前述の `CreateTable` SQL クエリをテーブル環境参照に対して使用してデータを読み取る方法を示しています。

```
   table_env.execute_sql(create_table(input_table, input_stream, input_region, stream_initpos))
```

### ストリーミングデータの書き込み
<a name="how-python-programming-readwrite-write"></a>

次のコード例は、`CreateTable` 例の SQL クエリを使用して出力テーブル参照を作成する方法と、`StatementSet`を使用してテーブルを操作して宛先の Kinesis ストリームにデータを書き込む方法を示しています。

```
   table_result = table_env.execute_sql("INSERT INTO {0} SELECT * FROM {1}"
                       .format(output_table_name, input_table_name))
```

## ランタイムプロパティの読み取り
<a name="how-python-programming-properties"></a>

ランタイムプロパティを使用すると、アプリケーションコードを変更せずにアプリケーションを設定できます。

アプリケーションのアプリケーションプロパティは、Java アプリケーション向けの Apache Flink 用 Managed Service と同じ方法で指定します。ランタイムプロパティは次の方法で指定できます。
+ 「[CreateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplication.html)」アクションを使用。
+ 「[UpdateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html)」アクションを使用。
+ コンソールを使ってアプリケーションを設定します。

コード内でアプリケーション・プロパティを取得するには、Managed Service for Apache Flinkランタイムが作成する `application_properties.json` と呼ばれるjsonファイルを読み込みます。

以下のサンプルコードは、`application_properties.json` ファイルからのアプリケーションプロパティの読み取りの例です。

```
file_path = '/etc/flink/application_properties.json'
   if os.path.isfile(file_path):
       with open(file_path, 'r') as file:
           contents = file.read()
           properties = json.loads(contents)
```

次のユーザー定義関数のコード例は、アプリケーションプロパティオブジェクト:retrieves からプロパティグループを読み取る方法を示しています。

```
def property_map(properties, property_group_id):
   for prop in props:
       if prop["PropertyGroupId"] == property_group_id:
           return prop["PropertyMap"]
```

次のコード例は、前の例で返されたプロパティグループから INPUT\_STREAM\_KEY というプロパティを読み取る方法を示しています。

```
input_stream = input_property_map[INPUT_STREAM_KEY]
```

## アプリケーションのコードパッケージの作成
<a name="how-python-programming-package"></a>

Python アプリケーションを作成したら、コードファイルと依存関係を zip ファイルにバンドルします。

zip ファイルには `main` メソッドを含む Python スクリプトが含まれている必要があり、オプションで以下を含めることができます。
+ その他の Python コードファイル
+ JAR ファイル内のユーザー定義 Java コード
+ JAR ファイル内の Java ライブラリ

**注記**  
アプリケーションの ZIP ファイルには、アプリケーションの依存関係がすべて含まれている必要があります。アプリケーションの他のソースからのライブラリを参照することはできません。