パイプラインで from_json を使用してスキーマを推論および進化する

Lakeflow パイプラインでは、 from_json SQL 関数は、明示的なスキーマを指定しなくても、JSON BLOB のスキーマを自動的に推論および進化させることができます。

パイプラインでの from_json のしくみ

from_json SQL 関数は、JSON 文字列列を解析し、構造体値を返します。 パイプラインの外部で使用する場合は、 schema 引数を使用して、戻り値のスキーマを明示的に指定する必要があります。 パイプラインで使用する場合は、返される値のスキーマを自動的に管理するスキーマの推論と進化を有効にすることができます。 この機能により、初期セットアップ (特にスキーマが不明な場合) と、スキーマが頻繁に変更される場合の継続的な操作の両方が簡略化されます。 自動ローダー、Kafka、Kinesis などのストリーミング データ ソースから任意の JSON BLOB を処理します。

具体的には、パイプラインで使用する場合、 from_json SQL 関数のスキーマ推論と進化は次のことができます。

• 受信 JSON レコード内の新しいフィールドを検出する (入れ子になった JSON オブジェクトを含む)
• フィールド型を推論し、適切な Spark データ型にマップする
• 新しいフィールドに対応するようにスキーマを自動的に進化させる
• 現在のスキーマに準拠していないデータを自動的に処理する

構文: スキーマを自動的に推論して進化させます

パイプラインで from_json を使用してスキーマ推論を有効にするには、スキーマを NULL に設定し、 schemaLocationKey オプションを指定します。 これにより、スキーマを推論して追跡できます。

SQL

from_json(jsonStr, NULL, map("schemaLocationKey", "<uniqueKey>” [, otherOptions]))

Python

from_json(jsonStr, None, {"schemaLocationKey": "<uniqueKey>”[, otherOptions]})

クエリには複数の from_json 式を含めることができますが、各式には一意の schemaLocationKeyが必要です。 schemaLocationKeyは、パイプラインごとに一意である必要もあります。

SQL

SELECT
  value,
  from_json(value, NULL, map('schemaLocationKey', 'keyX')) parsedX,
  from_json(value, NULL, map('schemaLocationKey', 'keyY')) parsedY,
FROM STREAM READ_FILES('/databricks-datasets/nyctaxi/sample/json/', format => 'text')

Python

(spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "text")
    .load("/databricks-datasets/nyctaxi/sample/json/")
    .select(
      col("value"),
      from_json(col("value"), None, {"schemaLocationKey": "keyX"}).alias("parsedX"),
      from_json(col("value"), None, {"schemaLocationKey": "keyY"}).alias("parsedY"))
)

構文: 固定スキーマ

代わりに特定のスキーマを適用する場合は、次の from_json 構文を使用して、そのスキーマを使用して JSON 文字列を解析できます。

from_json(jsonStr, schema, [, options])

この構文は、パイプラインを含む任意のAzure Databricks環境で使用できます。 詳細については、こちらを参照してください。

スキーマ推論

from_json JSON データ列の最初のバッチからスキーマを推論し、その schemaLocationKey によって内部的にインデックスを作成します (必須)。

JSON 文字列が単一のオブジェクト ( {"id": 123, "name": "John"} など) の場合、 from_json は STRUCT 型のスキーマを推論し、フィールドの一覧に rescuedDataColumn を追加します。

STRUCT<id LONG, name STRING, _rescued_data STRING>

ただし、JSON 文字列に最上位の配列 ( ["id": 123, "name": "John"] など) がある場合、 from_json は ARRAY を STRUCT でラップします。 この方法では、推論されたスキーマと互換性のないデータを取得できます。 配列値を下流の個別の行に 分解 するオプションがあります。

STRUCT<value ARRAY<id LONG, name STRING>, _rescued_data STRING>

スキーマ ヒントを使用してスキーマ推論をオーバーライドする

オプションでschemaHintsを指定して、from_jsonが列の型を推測する方法に影響を与えることができます。 これは、列が特定のデータ型であることがわかっている場合や、より一般的なデータ型 (整数ではなく double など) を選択する場合に役立ちます。 SQL スキーマ仕様の構文を使用して、列データ型に任意の数のヒントを指定できます。 スキーマ ヒントのセマンティクスは、自動ローダー スキーマ ヒントのセマンティクスと同じです。 例えば次が挙げられます。

SELECT
-- The JSON `{"a": 1}` will treat `a` as a BIGINT
from_json(data, NULL, map('schemaLocationKey', 'w', 'schemaHints', '')),
-- The JSON `{"a": 1}` will treat `a` as a STRING
from_json(data, NULL, map('schemaLocationKey', 'x', 'schemaHints', 'a STRING')),
-- The JSON `{"a": {"b": 1}}` will treat `a` as a MAP<STRING, BIGINT>
from_json(data, NULL, map('schemaLocationKey', 'y', 'schemaHints', 'a MAP<STRING, BIGINT'>)),
-- The JSON `{"a": {"b": 1}}` will treat `a` as a STRING
from_json(data, NULL, map('schemaLocationKey', 'z', 'schemaHints', 'a STRING')),
FROM STREAM READ_FILES(...)

JSON 文字列に最上位の ARRAY が含まれている場合は、STRUCT にラップされます。 このような場合、ラップされたSTRUCTではなく、ARRAYスキーマにスキーマヒントが適用されます。 たとえば、次のような最上位の配列を含む JSON 文字列を考えてみましょう。

[{"id": 123, "name": "John"}]

推論された ARRAY スキーマは、STRUCT でラップされます。

STRUCT<value ARRAY<id LONG, name STRING>, _rescued_data STRING>

idのデータ型を変更するには、スキーマ ヒントを文字列element.id指定します。 DOUBLE 型の新しい列を追加するには、DOUBLE element.new_col 指定します。 これらのヒントにより、最上位の JSON 配列のスキーマは次のようになります。

struct<value array<id STRING, name STRING, new_col DOUBLE>, _rescued_data STRING>

を使用してスキーマを進化させます schemaEvolutionMode

from_json は、データを処理しながら新しい列の追加を検出します。 from_jsonが新しいフィールドを検出すると、新しい列をスキーマの末尾にマージすることで、推論されたスキーマを最新のスキーマで更新します。 既存の列のデータ型は変更されません。 スキーマの更新後、更新されたスキーマを使用してパイプラインが自動的に再起動されます。

from_json では、オプションの schemaEvolutionMode 設定を使用して設定したスキーマの進化に対して、次のモードがサポートされています。 これらのモードは 自動ローダーと一致します。

例えば次が挙げられます。

SELECT
-- If a new column appears, the pipeline will automatically add it to the schema:
from_json(a, NULL, map('schemaLocationKey', 'w', 'schemaEvolutionMode', 'addNewColumns')),
-- If a new column appears, the pipeline will add it to the rescued data column:
from_json(b, NULL, map('schemaLocationKey', 'x', 'schemaEvolutionMode', 'rescue')),
-- If a new column appears, the pipeline will ignore it:
from_json(c, NULL, map('schemaLocationKey', 'y', 'schemaEvolutionMode', 'none')),
-- If a new column appears, the pipeline will fail:
from_json(d, NULL, map('schemaLocationKey', 'z', 'schemaEvolutionMode', 'failOnNewColumns')),
FROM STREAM READ_FILES(...)

復旧されたデータ列

復旧されたデータ列は、 _rescued_dataとしてスキーマに自動的に追加されます。 rescuedDataColumn オプションを設定すると、列の名前を変更できます。 例えば次が挙げられます。

from_json(jsonStr, None, {"schemaLocationKey": "keyX", "rescuedDataColumn": "my_rescued_data"})

復旧されたデータ列を使用することを選択すると、推論されたスキーマと一致しない列は削除されるのではなく、復旧されます。 これは、データ型が一致しない、スキーマに列がない、列名の大文字と小文字の違いが原因で発生する可能性があります。

破損したレコードを処理する

不正な形式で解析できないレコードを格納するには、次の例のようにスキーマ ヒントを設定して _corrupt_record 列を追加します。

CREATE STREAMING TABLE bronze AS
  SELECT
    from_json(value, NULL,
      map('schemaLocationKey', 'nycTaxi',
          'schemaHints', '_corrupt_record STRING',
          'columnNameOfCorruptRecord', '_corrupt_record')) jsonCol
  FROM STREAM READ_FILES('/databricks-datasets/nyctaxi/sample/json/', format => 'text')

破損したレコード列の名前を変更するには、 columnNameOfCorruptRecord オプションを設定します。

JSON パーサーでは、破損したレコードを処理するための 3 つのモードがサポートされています。

from_json出力のフィールドを参照する

from_json は、パイプラインの実行中にスキーマを推論します。 from_json関数が少なくとも 1 回正常に実行される前にダウンストリーム クエリがfrom_json フィールドを参照している場合、フィールドは解決せず、クエリはスキップされます。 次の例では、シルバー テーブル クエリの分析は、ブロンズ クエリの from_json 関数が実行され、スキーマが推論されるまでスキップされます。

CREATE STREAMING TABLE bronze AS
  SELECT
    from_json(value, NULL, map('schemaLocationKey', 'nycTaxi')) jsonCol
  FROM STREAM READ_FILES('/databricks-datasets/nyctaxi/sample/json/', format => 'text')

CREATE STREAMING TABLE silver AS
  SELECT jsonCol.VendorID, jsonCol.total_amount
  FROM bronze

from_json関数と推論するフィールドが同じクエリで参照されている場合、次の例のように分析が失敗する可能性があります。

CREATE STREAMING TABLE bronze AS
  SELECT
    from_json(value, NULL, map('schemaLocationKey', 'nycTaxi')) jsonCol
  FROM STREAM READ_FILES('/databricks-datasets/nyctaxi/sample/json/', format => 'text')
  WHERE jsonCol.total_amount > 100.0

これを修正するには、from_json フィールドへの参照をダウンストリーム クエリ (上のブロンズ/シルバーの例など) に移動します。または、参照されるschemaHintsフィールドを含むfrom_jsonを指定することもできます。 例えば次が挙げられます。

CREATE STREAMING TABLE bronze AS
  SELECT
    from_json(value, NULL, map('schemaLocationKey', 'nycTaxi', 'schemaHints', 'total_amount DOUBLE')) jsonCol
  FROM STREAM READ_FILES('/databricks-datasets/nyctaxi/sample/json/', format => 'text')
  WHERE jsonCol.total_amount > 100.0

例: スキーマを自動的に推論および進化する

このセクションでは、パイプラインで from_json を使用してスキーマの自動推論と進化を有効にするコード例を示します。

クラウド オブジェクト ストレージからストリーミング テーブルを作成する

次の例では、 read_files 構文を使用して、クラウド オブジェクト ストレージからストリーミング テーブルを作成します。

SQL

CREATE STREAMING TABLE bronze AS
  SELECT
    from_json(value, NULL, map('schemaLocationKey', 'nycTaxi')) jsonCol
  FROM STREAM READ_FILES('/databricks-datasets/nyctaxi/sample/json/', format => 'text')

Python

@dp.table(comment="from_json autoloader example")
def bronze():
  return (
    spark.readStream
         .format("cloudFiles")
         .option("cloudFiles.format", "text")
         .load("/databricks-datasets/nyctaxi/sample/json/")
         .select(from_json(col("value"), None, {"schemaLocationKey": "nycTaxi"}).alias("jsonCol"))
)

Kafka からストリーミング テーブルを作成する

次の例では、 read_kafka 構文を使用して Kafka からストリーミング テーブルを作成します。

SQL

CREATE STREAMING TABLE bronze AS
  SELECT
    value,
    from_json(value, NULL, map('schemaLocationKey', 'keyX')) jsonCol,
  FROM READ_KAFKA(
    bootstrapSevers => '<server:ip>',
    subscribe => 'events',
    "startingOffsets", "latest"
)

Python

@dp.table(comment="from_json kafka example")
def bronze():
  return (
    spark.readStream
         .format("kafka")
         .option("kafka.bootstrap.servers", "<server:ip>")
         .option("subscribe", "<topic>")
         .option("startingOffsets", "latest")
         .load()
         .select(col(“value”), from_json(col(“value”), None, {"schemaLocationKey": "keyX"}).alias("jsonCol"))
)

例: 固定スキーマ

固定スキーマで from_json を使用するコード例については、 from_json 関数を参照してください。

FAQs

このセクションでは、 from_json 関数でのスキーマ推論と進化のサポートに関してよく寄せられる質問に回答します。

from_jsonとparse_jsonの違いは何ですか?

parse_json関数は、JSON 文字列からVARIANT値を返します。

VARIANT は、半構造化データを柔軟かつ効率的に格納する方法を提供します。 これにより、厳密な型を完全に廃止することで、スキーマの推論と進化が回避されます。 ただし、書き込み時にスキーマを適用する場合 (たとえば、比較的厳密なスキーマがあるため)、 from_json の方が適している可能性があります。

次の表では、 from_json と parse_jsonの違いについて説明します。

パイプラインの外部 from_json スキーマ推論と進化構文を使用できますか?

いいえ。パイプラインの外部でスキーマ推論と進化構文 from_json 使用することはできません。

from_jsonによって推論されたスキーマにアクセスするにはどうすればよいですか?

ターゲット ストリーミング テーブルのスキーマを表示します。

スキーマ from_json 渡すことも、進化を行うこともできますか?

いいえ、スキーマ from_json 渡したり、進化を行うことはできません。 ただし、スキーマ ヒントを指定して、 from_jsonによって推論された一部またはすべてのフィールドをオーバーライドできます。

テーブルが完全に更新された場合、スキーマはどうなりますか?

テーブルに関連付けられているスキーマの場所がクリアされ、スキーマが最初から再推論されます。