Flink、Kafka、OSS、Elasticsearch コネクタでの JSON データ解析 - Flink

JSON 形式は、JSON データの読み取りおよび書き込みをサポートします。JSON 構造は、テーブルスキーマから自動的に推論されます。Realtime Compute for Apache Flink では、jackson databind API を使用して JSON データを解析および生成します。

以下のコネクタで JSON 形式がサポートされています：

例

以下の例では、JSON 形式でデータを読み書きする Kafka テーブルを作成します。

CREATE TABLE Orders (
  orderId    INT,
  product    STRING,
  orderInfo  MAP<STRING, STRING>,
  orderTime  TIMESTAMP(3),
  WATERMARK FOR orderTime AS orderTime - INTERVAL '5' SECOND
) WITH (
  'connector'                  = 'kafka',
  'topic'                      = 'test-topic',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format'                     = 'json',
  'json.fail-on-missing-field' = 'false',
  'json.ignore-parse-errors'   = 'true'
);

パラメーター

パラメーター	必須	デフォルト値	型	説明
`format`	はい	（なし）	STRING	使用するフォーマットを指定します。`json` を設定します。
`json.fail-on-missing-field`	いいえ	false	BOOLEAN	フィールドの欠落時にそのフィールドまたは行をスキップするかどうかを指定します。`true`：フィールドまたは行をスキップします。`false`：エラーを返し、デプロイメントを失敗させます。
`json.ignore-parse-errors`	いいえ	false	BOOLEAN	解析エラー発生時にそのフィールドまたは行をスキップするかどうかを指定します。`true`：フィールドまたは行をスキップします。`false`：エラーを返し、デプロイメントを失敗させます。
`json.timestamp-format.standard`	いいえ	SQL	STRING	入力および出力におけるタイムスタンプのフォーマットを指定します。`SQL`：`yyyy-MM-dd HH:mm:ss.s{precision}`（例：`2020-12-30 12:13:14.123`）。`ISO-8601`：`yyyy-MM-ddTHH:mm:ss.s{precision}`（例：`2020-12-30T12:13:14.123`）。
`json.map-null-key.mode`	いいえ	FAIL	STRING	マップ内の null キーをどのように処理するかを指定します。`FAIL`：エラーを返します。`DROP`：エントリを破棄します。`LITERAL`：`json.map-null-key.literal` で指定された文字列で null キーを置き換えます。
`json.map-null-key.literal`	いいえ	null	STRING	`json.map-null-key.mode` が `LITERAL` に設定されている場合に、null マップキーを置き換えるために使用される文字列定数です。
`json.encode.decimal-as-plain-number`	いいえ	false	BOOLEAN	10 進数値を通常表記でエンコードするかどうかを指定します。`true`：通常表記を使用（例：`0.000000027`）。`false`：指数表記を使用（例：`2.7E-8`）。
`json.write-null-properties`	いいえ	true	BOOLEAN	JSON 出力に null 列値を書き込むかどうかを指定します。`true`：null 値を書き込みます。`false`：null フィールドを省略します。Ververica Runtime (VVR) 8.0.6 以降が必要です。

データ型マッピング

Flink SQL 型	JSON 型
CHAR / VARCHAR / STRING	STRING
BOOLEAN	BOOLEAN
BINARY / VARBINARY	STRING（base64 エンコード）
DECIMAL	NUMBER
TINYINT	NUMBER
SMALLINT	NUMBER
INT	NUMBER
BIGINT	NUMBER
FLOAT	NUMBER
DOUBLE	NUMBER
DATE	STRING（`date` 形式）
TIME	STRING（`time` 形式）
TIMESTAMP	STRING（`date-time` 形式）
TIMESTAMP_WITH_LOCAL_TIME_ZONE	STRING（`date-time` 形式、UTC）
INTERVAL	NUMBER
ARRAY	ARRAY
MAP / MULTISET	OBJECT
ROW	OBJECT

制限事項

JSON 形式を介して OSS に書き込まれたデータは、JSON ファイルとして保存できません。詳細については、「FLINK-30635」をご参照ください。