Avro

Bases: ReadWriteFileFormat

Avro file format.

Based on Spark Avro file format.

Supports reading/writing files with .avro extension.

Version compatibility

• Spark versions: 2.4.x - 3.5.x
• Java versions: 8 - 20

See documentation from link above.

Added in 0.9.0

Examples

Note

You can pass any option mentioned in official documentation. Option names should be in camelCase!

The set of supported options depends on Spark version.

Reading filesWriting files

from pyspark.sql import SparkSession
from onetl.file.format import Avro

# Create Spark session with Avro package loaded
maven_packages = Avro.get_packages(spark_version="3.5.8")
spark = (
    SparkSession.builder.appName("spark-app-name")
    .config("spark.jars.packages", ",".join(maven_packages))
    .getOrCreate()
)

schema = {
    "type": "record",
    "name": "Person",
    "fields": [
        {"name": "name", "type": "string"},
        {"name": "age", "type": "int"},
    ],
}
avro = Avro(avroSchema=schema)  # or avroSchemaUrl=...

# Create Spark session with Avro package loaded
spark = ...

from onetl.file.format import Avro

schema = {
    "type": "record",
    "name": "Person",
    "fields": [
        {"name": "name", "type": "string"},
        {"name": "age", "type": "int"},
    ],
}
avro = Avro(
    avroSchema=schema,  # or avroSchemaUrl=...
    compression="snappy",
)

Source code in onetl/file/format/avro.py

@support_hooks
class Avro(ReadWriteFileFormat):
    """
    Avro file format. [![support hooks](https://img.shields.io/badge/%20-support%20hooks-blue)](/hooks/)

    Based on [Spark Avro](https://spark.apache.org/docs/latest/sql-data-sources-avro.html) file format.

    Supports reading/writing files with `.avro` extension.

    ??? note "Version compatibility"

        * Spark versions: 2.4.x - 3.5.x
        * Java versions: 8 - 20

        See documentation from link above.

    !!! success "Added in 0.9.0"

    Examples
    --------

    !!! note

        You can pass any option mentioned in
        [official documentation](https://spark.apache.org/docs/latest/sql-data-sources-avro.html).
        **Option names should be in** `camelCase`!

        The set of supported options depends on Spark version.

    === "Reading files"
        ```python
        from pyspark.sql import SparkSession
        from onetl.file.format import Avro

        # Create Spark session with Avro package loaded
        maven_packages = Avro.get_packages(spark_version="3.5.8")
        spark = (
            SparkSession.builder.appName("spark-app-name")
            .config("spark.jars.packages", ",".join(maven_packages))
            .getOrCreate()
        )

        schema = {
            "type": "record",
            "name": "Person",
            "fields": [
                {"name": "name", "type": "string"},
                {"name": "age", "type": "int"},
            ],
        }
        avro = Avro(avroSchema=schema)  # or avroSchemaUrl=...
        ```
    === "Writing files"
        ```python
        # Create Spark session with Avro package loaded
        spark = ...

        from onetl.file.format import Avro

        schema = {
            "type": "record",
            "name": "Person",
            "fields": [
                {"name": "name", "type": "string"},
                {"name": "age", "type": "int"},
            ],
        }
        avro = Avro(
            avroSchema=schema,  # or avroSchemaUrl=...
            compression="snappy",
        )
        ```
    """

    name: ClassVar[str] = "avro"

    schema_dict: Optional[dict] = Field(default=None, alias="avroSchema")
    """
    Avro schema in JSON format representation.

    ```python
    avro = Avro(
        avroSchema={
            "type": "record",
            "name": "Person",
            "fields": [
                {"name": "name", "type": "string"},
                {"name": "age", "type": "int"},
            ],
        },
    )
    ```
    If set, all records should match this schema.

    !!! warning

        Mutually exclusive with [schema_url][].
    """

    schema_url: Optional[str] = Field(default=None, alias="avroSchemaUrl")
    """
    URL to Avro schema in JSON format. Usually points to Schema Registry, like:

    ```python
    schema_registry = "http://some.schema.registry.domain"
    name = "MyAwesomeSchema"
    version = "latest"

    schema_url = f"{schema_registry}/subjects/{name}/versions/{version}/schema"
    avro = Avro(avroSchemaUrl=schema_url)
    ```
    If set, schema is fetched before any records are parsed, so all records should match this schema.

    !!! warning

        Mutually exclusive with [schema_dict][].
    """

    recordName: Optional[str] = None
    """
    Record name in written Avro schema.
    Default is `topLevelRecord`.

    !!! note

        Used only for writing files and by [serialize_column][].
    """

    recordNamespace: Optional[str] = None
    """
    Record namespace in written Avro schema. Default is not set.

    !!! note

        Used only for writing files and by [serialize_column][].
    """

    compression: Union[str, Literal["uncompressed", "snappy", "deflate", "bzip2", "xz", "zstandard"], None] = None
    """
    Compression codec.
    By default, Spark config value `spark.sql.avro.compression.codec ` (`snappy`) is used.

    !!! note

        Used only for writing files. Ignored by [serialize_column][].
    """

    mode: Optional[Literal["PERMISSIVE", "FAILFAST"]] = None
    """
    How to handle parsing errors:
      * `PERMISSIVE` - set field value as `null`.
      * `FAILFAST` - throw an error immediately.

    Default is `FAILFAST`.

    !!! note

        Used only by [parse_column][] method.
    """

    datetimeRebaseMode: Optional[Literal["CORRECTED", "LEGACY", "EXCEPTION"]] = None
    """
    While converting dates/timestamps from Julian to Proleptic Gregorian calendar, handle value ambiguity:
      * `EXCEPTION` - fail if ancient dates/timestamps are ambiguous between the two calendars.
      * `CORRECTED` - load dates/timestamps without as-is.
      * `LEGACY` - rebase ancient dates/timestamps from the Julian to Proleptic Gregorian calendar.

    By default, Spark config value `spark.sql.avro.datetimeRebaseModeInRead` (`CORRECTED`) is used.

    !!! note

        Used only for reading files and by [parse_column][].
    """

    positionalFieldMatching: Optional[bool] = None
    """
    If `True`, match Avro schema field and DataFrame column by position.
    If `False`, match by name.

    Default is `False`.
    """

    enableStableIdentifiersForUnionType: Optional[bool] = None
    """
    Avro schema may contain union types, which are not supported by Spark.
    Different variants of union are split to separated DataFrame columns with respective type.

    If option value is `True`, DataFrame column names are based on Avro variant names,
    e.g. `member_int`, `member_string`.
    If `False`, DataFrame column names are generated using field position, e.g. `member0`, `member1`.

    Default is `False`.

    !!! note

        Used only for reading files and by [parse_column][].
    """

    class Config:
        known_options: frozenset[str] = frozenset()
        prohibited_options = PROHIBITED_OPTIONS
        extra = "allow"

    @slot
    @classmethod
    def get_packages(
        cls,
        spark_version: str,
        scala_version: str | None = None,
    ) -> list[str]:
        """
        Get package names to be downloaded by Spark. [![support hooks](https://img.shields.io/badge/%20-support%20hooks-blue)](/hooks/)

        See [Maven package index](https://mvnrepository.com/artifact/org.apache.spark/spark-avro)
        for all available packages.

        !!! success "Added in 0.9.0"

        Parameters
        ----------
        spark_version : str
            Spark version in format `major.minor.patch`.

        scala_version : str, optional
            Scala version in format `major.minor`.

            If `None`, `spark_version` is used to determine Scala version.

        Examples
        --------

        ```python
        from onetl.file.format import Avro

        Avro.get_packages(spark_version="3.5.8")
        Avro.get_packages(spark_version="3.5.8", scala_version="2.12")
        ```
        """

        spark_ver = Version(spark_version).min_digits(3)
        scala_ver = Version(scala_version).min_digits(2) if scala_version else get_default_scala_version(spark_ver)
        return [f"org.apache.spark:spark-avro_{scala_ver.format('{0}.{1}')}:{spark_ver.format('{0}.{1}.{2}')}"]

    @slot
    def check_if_supported(self, spark: SparkSession) -> None:
        java_class = "org.apache.spark.sql.avro.AvroFileFormat"

        try:
            try_import_java_class(spark, java_class)
        except Exception as e:
            spark_version = get_spark_version(spark).format("{0}.{1}.{2}")
            msg = MISSING_JVM_CLASS_MSG.format(
                java_class=java_class,
                package_source=self.__class__.__name__,
                args=f"spark_version='{spark_version}'",
            )
            raise ValueError(msg) from e

    @slot
    def apply_to_reader(self, reader: DataFrameReader) -> DataFrameReader:
        options = self.dict(by_alias=True, exclude_none=True, exclude={"schema"})
        if self.schema_dict:
            options["avroSchema"] = json.dumps(self.schema_dict)
        return reader.format(self.name).options(**options)

    @slot
    def apply_to_writer(self, writer: DataFrameWriter) -> DataFrameWriter:
        options = self.dict(by_alias=True, exclude_none=True, exclude={"schema"})
        if self.schema_dict:
            options["avroSchema"] = json.dumps(self.schema_dict)
        return writer.format(self.name).options(**options)

    def parse_column(self, column: str | Column) -> Column:
        """
        Parses an Avro binary column into a structured Spark SQL column using Spark's
        [from_avro](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.avro.functions.from_avro.html) function,
        based on the schema provided within the class.

        !!! note

            Can be used only with Spark 3.x+

        !!! warning

            If `schema_url` is provided, `requests` library is used to fetch the schema from the URL.
            It should be installed manually, like this:

            ```bash
            pip install requests
            ```
        !!! success "Added in 0.11.0"

        Parameters
        ----------
        column : str | Column
            The name of the column or the column object containing Avro bytes to deserialize.
            Schema should match the provided Avro schema.

        Returns
        -------
        pyspark.sql.Column
            Column with deserialized data. Schema is matching the provided Avro schema.
            Column name is the same as input column.

        Raises
        ------
        ValueError
            If neither `avroSchema` nor `avroSchemaUrl` are defined.
        ImportError
            If `schema_url` is used and the `requests` library is not installed.

        Examples
        --------

        ```python
        >>> from pyspark.sql.functions import decode
        >>> from onetl.file.format import Avro
        >>> df.show()
        +----+----------------------+----------+---------+------+-----------------------+-------------+
        |key |value                 |topic     |partition|offset|timestamp              |timestampType|
        +----+----------------------+----------+---------+------+-----------------------+-------------+
        |[31]|[0A 41 6C 69 63 65 28]|topicAvro |0        |0     |2024-04-24 13:02:25.911|0            |
        |[32]|[06 42 6F 62 32]      |topicAvro |0        |1     |2024-04-24 13:02:25.922|0            |
        +----+----------------------+----------+---------+------+-----------------------+-------------+
        >>> df.printSchema()
        root
        |-- key: binary (nullable = true)
        |-- value: binary (nullable = true)
        |-- topic: string (nullable = true)
        |-- partition: integer (nullable = true)
        |-- offset: integer (nullable = true)
        |-- timestamp: timestamp (nullable = true)
        |-- timestampType: integer (nullable = true)
        >>> avro = Avro(
        ...     avroSchema={  # or avroSchemaUrl=...
        ...         "type": "record",
        ...         "name": "Person",
        ...         "fields": [
        ...             {"name": "name", "type": "string"},
        ...             {"name": "age", "type": "int"},
        ...         ],
        ...     }
        ... )
        >>> parsed_df = df.select(decode("key", "UTF-8").alias("key"), avro.parse_column("value"))
        >>> parsed_df.show(truncate=False)
        +---+-----------+
        |key|value      |
        +---+-----------+
        |1  |{Alice, 20}|
        |2  |{Bob, 25}  |
        +---+-----------+
        >>> parsed_df.printSchema()
        root
        |-- key: string (nullable = true)
        |-- value: struct (nullable = true)
        |    |-- name: string (nullable = true)
        |    |-- age: integer (nullable = true)
        ```
        """  # noqa: E501
        from pyspark.sql import Column, SparkSession
        from pyspark.sql.functions import col

        spark = cast("SparkSession", SparkSession._instantiatedSession)  # noqa: SLF001
        self.check_if_supported(spark)
        self._check_unsupported_parse_options()

        from pyspark.sql.avro.functions import from_avro

        if isinstance(column, Column):
            column_name = column._jc.toString()  # noqa: SLF001
        else:
            column_name, column = column, col(column).cast("binary")

        schema = self._get_schema_json()
        if not schema:
            msg = "Avro.parse_column can be used only with defined `avroSchema` or `avroSchemaUrl`"
            raise ValueError(msg)

        return from_avro(column, schema).alias(column_name)

    def serialize_column(self, column: str | Column) -> Column:
        """
        Serializes a structured Spark SQL column into an Avro binary column using Spark's
        [to_avro](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.avro.functions.to_avro.html#pyspark.sql.avro.functions.to_avro) function.

        !!! note

            Can be used only with Spark 3.x+

        !!! warning

            If `schema_url` is provided, `requests` library is used to fetch the schema from the URL.
            It should be installed manually, like this:

            ```bash
            pip install requests
            ```
        !!! success "Added in 0.11.0"

        Parameters
        ----------
        column : str | Column
            The name of the column or the column object containing the data to serialize to Avro format.

        Returns
        -------
        pyspark.sql.Column
            Column with binary Avro data. Column name is the same as input column.

        Raises
        ------
        ValueError
            If the Spark version is less than 3.x.
        ImportError
            If `schema_url` is used and the `requests` library is not installed.

        Examples
        --------

        ```python
        >>> from pyspark.sql.functions import decode
        >>> from onetl.file.format import Avro
        >>> df.show()
        +---+-----------+
        |key|value      |
        +---+-----------+
        |1  |{Alice, 20}|
        |2  |  {Bob, 25}|
        +---+-----------+
        >>> df.printSchema()
        root
        |-- key: string (nullable = true)
        |-- value: struct (nullable = true)
        |    |-- name: string (nullable = true)
        |    |-- age: integer (nullable = true)
        >>> # serializing data into Avro format
        >>> avro = Avro(
        ...     avroSchema={  # or avroSchemaUrl=...
        ...         "type": "record",
        ...         "name": "Person",
        ...         "fields": [
        ...             {"name": "name", "type": "string"},
        ...             {"name": "age", "type": "int"},
        ...         ],
        ...     }
        ... )
        >>> serialized_df = df.select("key", avro.serialize_column("value"))
        >>> serialized_df.show(truncate=False)
        +---+----------------------+
        |key|value                 |
        +---+----------------------+
        |  1|[0A 41 6C 69 63 65 28]|
        |  2|[06 42 6F 62 32]      |
        +---+----------------------+
        >>> serialized_df.printSchema()
        root
        |-- key: string (nullable = true)
        |-- value: binary (nullable = true)
        ```
        """  # noqa: E501
        from pyspark.sql import Column, SparkSession
        from pyspark.sql.functions import col

        spark = cast("SparkSession", SparkSession._instantiatedSession)  # noqa: SLF001
        self.check_if_supported(spark)
        self._check_unsupported_serialization_options()

        from pyspark.sql.avro.functions import to_avro

        if isinstance(column, Column):
            column_name = column._jc.toString()  # noqa: SLF001
        else:
            column_name, column = column, col(column)

        schema = self._get_schema_json()
        return to_avro(column, schema).alias(column_name)

    def __repr__(self):
        options_dict = self.dict(by_alias=True, exclude_none=True)
        options_kwargs = ", ".join(f"{k}={v!r}" for k, v in options_dict.items())
        return f"{self.__class__.__name__}({options_kwargs})"

    def _check_unsupported_parse_options(self):
        current_options = self.dict(by_alias=True, exclude_none=True)
        unsupported_options = current_options.keys() & PARSE_COLUMN_UNSUPPORTED_OPTIONS
        if unsupported_options:
            warnings.warn(
                f"Options `{sorted(unsupported_options)}` are set but not supported by `Avro.parse_column`.",
                UserWarning,
                stacklevel=2,
            )

    def _check_unsupported_serialization_options(self):
        current_options = self.dict(by_alias=True, exclude_none=True)
        unsupported_options = current_options.keys() & SERIALIZE_COLUMN_UNSUPPORTED_OPTIONS
        if unsupported_options:
            warnings.warn(
                f"Options `{sorted(unsupported_options)}` are set but not supported by `Avro.serialize_column`.",
                UserWarning,
                stacklevel=2,
            )

    @validator("schema_dict", pre=True)
    def _parse_schema_from_json(cls, value):
        if isinstance(value, (str, bytes)):
            return json.loads(value)
        return value

    @root_validator(pre=True)
    def _check_schema(cls, values):
        schema_dict = values.get("schema_dict")
        schema_url = values.get("schema_url")
        if schema_dict and schema_url:
            msg = "Parameters `avroSchema` and `avroSchemaUrl` are mutually exclusive."
            raise ValueError(msg)
        return values

    def _get_schema_json(self) -> str:
        if self.schema_dict:
            return json.dumps(self.schema_dict)
        if self.schema_url:
            try:
                import requests
            except ImportError as e:
                msg = (
                    "The 'requests' library is required to use 'schema_url' but is not installed. "
                    "Install it with 'pip install requests' or avoid using 'schema_url'."
                )
                raise ImportError(msg) from e
            else:
                response = requests.get(self.schema_url)  # noqa: S113
                return response.text
        else:
            return ""

schema_dict = Field(default=None, alias='avroSchema') class-attribute instance-attribute

Avro schema in JSON format representation.

avro = Avro(
    avroSchema={
        "type": "record",
        "name": "Person",
        "fields": [
            {"name": "name", "type": "string"},
            {"name": "age", "type": "int"},
        ],
    },
)

If set, all records should match this schema.

Warning

Mutually exclusive with [schema_url][].

schema_url = Field(default=None, alias='avroSchemaUrl') class-attribute instance-attribute

URL to Avro schema in JSON format. Usually points to Schema Registry, like:

schema_registry = "http://some.schema.registry.domain"
name = "MyAwesomeSchema"
version = "latest"

schema_url = f"{schema_registry}/subjects/{name}/versions/{version}/schema"
avro = Avro(avroSchemaUrl=schema_url)

If set, schema is fetched before any records are parsed, so all records should match this schema.

Warning

Mutually exclusive with [schema_dict][].

recordName = None class-attribute instance-attribute

Record name in written Avro schema. Default is topLevelRecord.

Note

Used only for writing files and by [serialize_column][].

recordNamespace = None class-attribute instance-attribute

Record namespace in written Avro schema. Default is not set.

Note

Used only for writing files and by [serialize_column][].

compression = None class-attribute instance-attribute

Compression codec. By default, Spark config value spark.sql.avro.compression.codec (snappy) is used.

Note

Used only for writing files. Ignored by [serialize_column][].

mode = None class-attribute instance-attribute

How to handle parsing errors

• PERMISSIVE - set field value as null.
• FAILFAST - throw an error immediately.

Default is FAILFAST.

Note

Used only by [parse_column][] method.

datetimeRebaseMode = None class-attribute instance-attribute

While converting dates/timestamps from Julian to Proleptic Gregorian calendar, handle value ambiguity: * EXCEPTION - fail if ancient dates/timestamps are ambiguous between the two calendars. * CORRECTED - load dates/timestamps without as-is. * LEGACY - rebase ancient dates/timestamps from the Julian to Proleptic Gregorian calendar.

By default, Spark config value spark.sql.avro.datetimeRebaseModeInRead (CORRECTED) is used.

Note

Used only for reading files and by [parse_column][].

positionalFieldMatching = None class-attribute instance-attribute

If True, match Avro schema field and DataFrame column by position. If False, match by name.

Default is False.

enableStableIdentifiersForUnionType = None class-attribute instance-attribute

Avro schema may contain union types, which are not supported by Spark. Different variants of union are split to separated DataFrame columns with respective type.

If option value is True, DataFrame column names are based on Avro variant names, e.g. member_int, member_string. If False, DataFrame column names are generated using field position, e.g. member0, member1.

Default is False.

Note

Used only for reading files and by [parse_column][].

get_packages(spark_version, scala_version=None) classmethod

Get package names to be downloaded by Spark.

See Maven package index for all available packages.

Added in 0.9.0

Parameters

spark_version : str Spark version in format major.minor.patch.

str, optional

Scala version in format major.minor.

If None, spark_version is used to determine Scala version.

Examples

from onetl.file.format import Avro

Avro.get_packages(spark_version="3.5.8")
Avro.get_packages(spark_version="3.5.8", scala_version="2.12")

Source code in onetl/file/format/avro.py

@slot
@classmethod
def get_packages(
    cls,
    spark_version: str,
    scala_version: str | None = None,
) -> list[str]:
    """
    Get package names to be downloaded by Spark. [![support hooks](https://img.shields.io/badge/%20-support%20hooks-blue)](/hooks/)

    See [Maven package index](https://mvnrepository.com/artifact/org.apache.spark/spark-avro)
    for all available packages.

    !!! success "Added in 0.9.0"

    Parameters
    ----------
    spark_version : str
        Spark version in format `major.minor.patch`.

    scala_version : str, optional
        Scala version in format `major.minor`.

        If `None`, `spark_version` is used to determine Scala version.

    Examples
    --------

    ```python
    from onetl.file.format import Avro

    Avro.get_packages(spark_version="3.5.8")
    Avro.get_packages(spark_version="3.5.8", scala_version="2.12")
    ```
    """

    spark_ver = Version(spark_version).min_digits(3)
    scala_ver = Version(scala_version).min_digits(2) if scala_version else get_default_scala_version(spark_ver)
    return [f"org.apache.spark:spark-avro_{scala_ver.format('{0}.{1}')}:{spark_ver.format('{0}.{1}.{2}')}"]

parse_column(column)

Parses an Avro binary column into a structured Spark SQL column using Spark's from_avro function, based on the schema provided within the class.

Note

Can be used only with Spark 3.x+

Warning

If schema_url is provided, requests library is used to fetch the schema from the URL. It should be installed manually, like this:

pip install requests

Added in 0.11.0

Parameters

column : str | Column The name of the column or the column object containing Avro bytes to deserialize. Schema should match the provided Avro schema.

Returns

pyspark.sql.Column Column with deserialized data. Schema is matching the provided Avro schema. Column name is the same as input column.

Raises

ValueError If neither avroSchema nor avroSchemaUrl are defined. ImportError If schema_url is used and the requests library is not installed.

Examples

>>> from pyspark.sql.functions import decode
>>> from onetl.file.format import Avro
>>> df.show()
+----+----------------------+----------+---------+------+-----------------------+-------------+
|key |value                 |topic     |partition|offset|timestamp              |timestampType|
+----+----------------------+----------+---------+------+-----------------------+-------------+
|[31]|[0A 41 6C 69 63 65 28]|topicAvro |0        |0     |2024-04-24 13:02:25.911|0            |
|[32]|[06 42 6F 62 32]      |topicAvro |0        |1     |2024-04-24 13:02:25.922|0            |
+----+----------------------+----------+---------+------+-----------------------+-------------+
>>> df.printSchema()
root
|-- key: binary (nullable = true)
|-- value: binary (nullable = true)
|-- topic: string (nullable = true)
|-- partition: integer (nullable = true)
|-- offset: integer (nullable = true)
|-- timestamp: timestamp (nullable = true)
|-- timestampType: integer (nullable = true)
>>> avro = Avro(
...     avroSchema={  # or avroSchemaUrl=...
...         "type": "record",
...         "name": "Person",
...         "fields": [
...             {"name": "name", "type": "string"},
...             {"name": "age", "type": "int"},
...         ],
...     }
... )
>>> parsed_df = df.select(decode("key", "UTF-8").alias("key"), avro.parse_column("value"))
>>> parsed_df.show(truncate=False)
+---+-----------+
|key|value      |
+---+-----------+
|1  |{Alice, 20}|
|2  |{Bob, 25}  |
+---+-----------+
>>> parsed_df.printSchema()
root
|-- key: string (nullable = true)
|-- value: struct (nullable = true)
|    |-- name: string (nullable = true)
|    |-- age: integer (nullable = true)

Source code in onetl/file/format/avro.py

def parse_column(self, column: str | Column) -> Column:
    """
    Parses an Avro binary column into a structured Spark SQL column using Spark's
    [from_avro](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.avro.functions.from_avro.html) function,
    based on the schema provided within the class.

    !!! note

        Can be used only with Spark 3.x+

    !!! warning

        If `schema_url` is provided, `requests` library is used to fetch the schema from the URL.
        It should be installed manually, like this:

        ```bash
        pip install requests
        ```
    !!! success "Added in 0.11.0"

    Parameters
    ----------
    column : str | Column
        The name of the column or the column object containing Avro bytes to deserialize.
        Schema should match the provided Avro schema.

    Returns
    -------
    pyspark.sql.Column
        Column with deserialized data. Schema is matching the provided Avro schema.
        Column name is the same as input column.

    Raises
    ------
    ValueError
        If neither `avroSchema` nor `avroSchemaUrl` are defined.
    ImportError
        If `schema_url` is used and the `requests` library is not installed.

    Examples
    --------

    ```python
    >>> from pyspark.sql.functions import decode
    >>> from onetl.file.format import Avro
    >>> df.show()
    +----+----------------------+----------+---------+------+-----------------------+-------------+
    |key |value                 |topic     |partition|offset|timestamp              |timestampType|
    +----+----------------------+----------+---------+------+-----------------------+-------------+
    |[31]|[0A 41 6C 69 63 65 28]|topicAvro |0        |0     |2024-04-24 13:02:25.911|0            |
    |[32]|[06 42 6F 62 32]      |topicAvro |0        |1     |2024-04-24 13:02:25.922|0            |
    +----+----------------------+----------+---------+------+-----------------------+-------------+
    >>> df.printSchema()
    root
    |-- key: binary (nullable = true)
    |-- value: binary (nullable = true)
    |-- topic: string (nullable = true)
    |-- partition: integer (nullable = true)
    |-- offset: integer (nullable = true)
    |-- timestamp: timestamp (nullable = true)
    |-- timestampType: integer (nullable = true)
    >>> avro = Avro(
    ...     avroSchema={  # or avroSchemaUrl=...
    ...         "type": "record",
    ...         "name": "Person",
    ...         "fields": [
    ...             {"name": "name", "type": "string"},
    ...             {"name": "age", "type": "int"},
    ...         ],
    ...     }
    ... )
    >>> parsed_df = df.select(decode("key", "UTF-8").alias("key"), avro.parse_column("value"))
    >>> parsed_df.show(truncate=False)
    +---+-----------+
    |key|value      |
    +---+-----------+
    |1  |{Alice, 20}|
    |2  |{Bob, 25}  |
    +---+-----------+
    >>> parsed_df.printSchema()
    root
    |-- key: string (nullable = true)
    |-- value: struct (nullable = true)
    |    |-- name: string (nullable = true)
    |    |-- age: integer (nullable = true)
    ```
    """  # noqa: E501
    from pyspark.sql import Column, SparkSession
    from pyspark.sql.functions import col

    spark = cast("SparkSession", SparkSession._instantiatedSession)  # noqa: SLF001
    self.check_if_supported(spark)
    self._check_unsupported_parse_options()

    from pyspark.sql.avro.functions import from_avro

    if isinstance(column, Column):
        column_name = column._jc.toString()  # noqa: SLF001
    else:
        column_name, column = column, col(column).cast("binary")

    schema = self._get_schema_json()
    if not schema:
        msg = "Avro.parse_column can be used only with defined `avroSchema` or `avroSchemaUrl`"
        raise ValueError(msg)

    return from_avro(column, schema).alias(column_name)

serialize_column(column)

Serializes a structured Spark SQL column into an Avro binary column using Spark's to_avro function.

Note

Can be used only with Spark 3.x+

Warning

If schema_url is provided, requests library is used to fetch the schema from the URL. It should be installed manually, like this:

pip install requests

Added in 0.11.0

Parameters

column : str | Column The name of the column or the column object containing the data to serialize to Avro format.

Returns

pyspark.sql.Column Column with binary Avro data. Column name is the same as input column.

Raises

ValueError If the Spark version is less than 3.x. ImportError If schema_url is used and the requests library is not installed.

Examples

>>> from pyspark.sql.functions import decode
>>> from onetl.file.format import Avro
>>> df.show()
+---+-----------+
|key|value      |
+---+-----------+
|1  |{Alice, 20}|
|2  |  {Bob, 25}|
+---+-----------+
>>> df.printSchema()
root
|-- key: string (nullable = true)
|-- value: struct (nullable = true)
|    |-- name: string (nullable = true)
|    |-- age: integer (nullable = true)
>>> # serializing data into Avro format
>>> avro = Avro(
...     avroSchema={  # or avroSchemaUrl=...
...         "type": "record",
...         "name": "Person",
...         "fields": [
...             {"name": "name", "type": "string"},
...             {"name": "age", "type": "int"},
...         ],
...     }
... )
>>> serialized_df = df.select("key", avro.serialize_column("value"))
>>> serialized_df.show(truncate=False)
+---+----------------------+
|key|value                 |
+---+----------------------+
|  1|[0A 41 6C 69 63 65 28]|
|  2|[06 42 6F 62 32]      |
+---+----------------------+
>>> serialized_df.printSchema()
root
|-- key: string (nullable = true)
|-- value: binary (nullable = true)

Source code in onetl/file/format/avro.py

def serialize_column(self, column: str | Column) -> Column:
    """
    Serializes a structured Spark SQL column into an Avro binary column using Spark's
    [to_avro](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.avro.functions.to_avro.html#pyspark.sql.avro.functions.to_avro) function.

    !!! note

        Can be used only with Spark 3.x+

    !!! warning

        If `schema_url` is provided, `requests` library is used to fetch the schema from the URL.
        It should be installed manually, like this:

        ```bash
        pip install requests
        ```
    !!! success "Added in 0.11.0"

    Parameters
    ----------
    column : str | Column
        The name of the column or the column object containing the data to serialize to Avro format.

    Returns
    -------
    pyspark.sql.Column
        Column with binary Avro data. Column name is the same as input column.

    Raises
    ------
    ValueError
        If the Spark version is less than 3.x.
    ImportError
        If `schema_url` is used and the `requests` library is not installed.

    Examples
    --------

    ```python
    >>> from pyspark.sql.functions import decode
    >>> from onetl.file.format import Avro
    >>> df.show()
    +---+-----------+
    |key|value      |
    +---+-----------+
    |1  |{Alice, 20}|
    |2  |  {Bob, 25}|
    +---+-----------+
    >>> df.printSchema()
    root
    |-- key: string (nullable = true)
    |-- value: struct (nullable = true)
    |    |-- name: string (nullable = true)
    |    |-- age: integer (nullable = true)
    >>> # serializing data into Avro format
    >>> avro = Avro(
    ...     avroSchema={  # or avroSchemaUrl=...
    ...         "type": "record",
    ...         "name": "Person",
    ...         "fields": [
    ...             {"name": "name", "type": "string"},
    ...             {"name": "age", "type": "int"},
    ...         ],
    ...     }
    ... )
    >>> serialized_df = df.select("key", avro.serialize_column("value"))
    >>> serialized_df.show(truncate=False)
    +---+----------------------+
    |key|value                 |
    +---+----------------------+
    |  1|[0A 41 6C 69 63 65 28]|
    |  2|[06 42 6F 62 32]      |
    +---+----------------------+
    >>> serialized_df.printSchema()
    root
    |-- key: string (nullable = true)
    |-- value: binary (nullable = true)
    ```
    """  # noqa: E501
    from pyspark.sql import Column, SparkSession
    from pyspark.sql.functions import col

    spark = cast("SparkSession", SparkSession._instantiatedSession)  # noqa: SLF001
    self.check_if_supported(spark)
    self._check_unsupported_serialization_options()

    from pyspark.sql.avro.functions import to_avro

    if isinstance(column, Column):
        column_name = column._jc.toString()  # noqa: SLF001
    else:
        column_name, column = column, col(column)

    schema = self._get_schema_json()
    return to_avro(column, schema).alias(column_name)