from_avro

Avro形式のバイナリ列を、対応するCatalyst値に変換します。指定されたスキーマは読み取ったデータと一致していなければなりません。一致しない場合、動作は未定義となり、失敗するか、任意の結果が返される可能性があります。

jsonFormatSchemaが指定されておらず、 subjectとschemaRegistryAddressの両方が指定されている場合、この関数はスキーマレジストリAvro形式のバイナリ列を対応するCatalyst値に変換します。

構文

Python

from pyspark.sql.avro.functions import from_avro

from_avro(data, jsonFormatSchema=None, options=None, subject=None, schemaRegistryAddress=None)

パラメーター

パラメーター	Type	説明
data	pyspark.sql.Column またはstr	Avroエンコードされたデータを含むバイナリ列。
jsonFormatSchema	文字列、オプション	JSON文字列形式のAvroスキーマ。
options	辞書、オプション	Avroレコードの解析方法を制御するオプションと、スキーマレジストリクライアントの設定。
subject	文字列、オプション	スキーマレジストリにおける、データが属する主体。
schemaRegistryAddress	文字列、オプション	スキーマレジストリのアドレス（ホスト名とポート番号）。

パラメーター	Type	説明
data	pyspark.sql.Column またはstr	Avroエンコードされたデータを含むバイナリ列。
jsonFormatSchema	文字列、オプション	JSON文字列形式のAvroスキーマ。
options	辞書、オプション	Avroレコードの解析方法を制御するオプションと、スキーマレジストリクライアントの設定。
subject	文字列、オプション	スキーマレジストリにおける、データが属する主体。
schemaRegistryAddress	文字列、オプション	スキーマレジストリのアドレス（ホスト名とポート番号）。

オプション

オプション	値	説明
mode	FAILFAST, PERMISSIVE	エラー処理モード。デフォルト: FAILFAST 。PERMISSIVEモードでは、破損したレコードはエラーを発生させる代わりにNULLに設定されます。
compression	uncompressed、snappy、deflate、bzip2、xz、 zstandard	Avroデータをエンコードするための圧縮コーデック。
avroSchemaEvolutionMode	none, restart	スキーマ進化モード。デフォルト: none 。restartに設定すると、スキーマが変更されたときにクエリはUnknownFieldExceptionをスローします。新しいスキーマを使用するには、ジョブを再起動してください。from_avro でスキーマ進化モードを使用する方法を参照してください。
recursiveFieldMaxDepth	範囲: -1 ～ 15	単一の再帰パスに沿った最大再帰深度。デフォルト値： -1 。これは再帰の深さを制限しません。 共有型が複数の異なるスキーマパスからアクセス可能な場合、このオプションは深さを1つのパスのみに制限するため、スキーマ展開によってドライバがメモリ不足になることがあります。 回避策： ドライバーのサイズを大きくしてください。 再帰参照をリーフ型に置き換えることで、スキーマを簡素化します。 ペイロードをVARIANT型で解析します。半構造化バリアントタイプとしてデータを取り込むを参照してください。

オプション	値	説明
mode	FAILFAST, PERMISSIVE	エラー処理モード。デフォルト: FAILFAST 。PERMISSIVEモードでは、破損したレコードはエラーを発生させる代わりにNULLに設定されます。
compression	uncompressed、snappy、deflate、bzip2、xz、 zstandard	Avroデータをエンコードするための圧縮コーデック。
avroSchemaEvolutionMode	none, restart	スキーマ進化モード。デフォルト: none 。restartに設定すると、スキーマが変更されたときにクエリはUnknownFieldExceptionをスローします。新しいスキーマを使用するには、ジョブを再起動してください。from_avro でスキーマ進化モードを使用する方法を参照してください。
recursiveFieldMaxDepth	範囲: -1 ～ 15	単一の再帰パスに沿った最大再帰深度。デフォルト値： -1 。これは再帰の深さを制限しません。 共有型が複数の異なるスキーマパスからアクセス可能な場合、このオプションは深さを1つのパスのみに制限するため、スキーマ展開によってドライバがメモリ不足になることがあります。 回避策： ドライバーのサイズを大きくしてください。 再帰参照をリーフ型に置き換えることで、スキーマを簡素化します。 ペイロードをVARIANT型で解析します。半構造化バリアントタイプとしてデータを取り込むを参照してください。

戻り値

pyspark.sql.Column: 対応する触媒値として、逆シリアル化された Avro データを含む新しい列。

例

例1 ：JSONスキーマを使用してAvroバイナリ列を逆シリアル化する

Python

from pyspark.sql import Row
from pyspark.sql.avro.functions import from_avro, to_avro

data = [(1, Row(age=2, name='Alice'))]
df = spark.createDataFrame(data, ("key", "value"))
avro_df = df.select(to_avro(df.value).alias("avro"))
json_format_schema = '''{"type":"record","name":"topLevelRecord","fields":
    [{"name":"avro","type":[{"type":"record","name":"value",
    "namespace":"topLevelRecord","fields":[{"name":"age","type":["long","null"]},
    {"name":"name","type":["string","null"]}]},"null"]}]}'''
avro_df.select(from_avro(avro_df.avro, json_format_schema).alias("value")).show(truncate=False)

Output

+------------------+
|value             |
+------------------+
|{{2, Alice}}      |
+------------------+