Buffers de protocolo de leitura e gravação

O Azure Databricks fornece suporte nativo para serialização e desserialização entre estruturas do Apache Spark e buffers de protocolo (protobuf). O suporte ao Protobuf é implementado como um transformador Apache Spark DataFrame e pode ser usado com Streaming Estruturado ou para operações em lote.

Como desserializar e serializar buffers de protocolo

No Databricks Runtime 12.2 LTS e superior, você pode usar from_protobuf e to_protobuf funções para serializar e desserializar dados. A serialização Protobuf é comumente usada em cargas de trabalho de streaming.

A sintaxe básica para funções protobuf é semelhante para funções de leitura e gravação. Você deve importar essas funções antes de usar.

from_protobuf Converte uma coluna binária em uma struct e converte to_protobuf uma coluna struct em binary. Você deve fornecer um registro de esquema especificado com o options argumento ou um arquivo descritor identificado pelo descFilePath argumento.

Python

from_protobuf(data: 'ColumnOrName', messageName: Optional[str] = None, descFilePath: Optional[str] = None, options: Optional[Dict[str, str]] = None)

to_protobuf(data: 'ColumnOrName', messageName: Optional[str] = None, descFilePath: Optional[str] = None, options: Optional[Dict[str, str]] = None)

Scala

// While using with Schema registry:
from_protobuf(data: Column, options: Map[String, String])

// Or with Protobuf descriptor file:
from_protobuf(data: Column, messageName: String, descFilePath: String, options: Map[String, String])

// While using with Schema registry:
to_protobuf(data: Column, options: Map[String, String])

// Or with Protobuf descriptor file:
to_protobuf(data: Column, messageName: String, descFilePath: String, options: Map[String, String])

Os exemplos a seguir ilustram o processamento de registros de protobuf binário com from_protobuf() e a conversão do Spark SQL struct em protobuf binário com to_protobuf().

Use protobuf com o Confluent Schema Registry

O Azure Databricks dá suporte ao uso do Registro de Esquema Confluente para definir o Protobuf.

Python

from pyspark.sql.protobuf.functions import to_protobuf, from_protobuf

schema_registry_options = {
  "schema.registry.subject" : "app-events-value",
  "schema.registry.address" : "https://schema-registry:8081/"
}

# Convert binary Protobuf to SQL struct with from_protobuf():
proto_events_df = (
  input_df
    .select(
      from_protobuf("proto_bytes", options = schema_registry_options)
        .alias("proto_event")
    )
)

# Convert SQL struct to binary Protobuf with to_protobuf():
protobuf_binary_df = (
  proto_events_df
    .selectExpr("struct(name, id, context) as event")
    .select(
      to_protobuf("event", options = schema_registry_options)
        .alias("proto_bytes")
    )
)

Scala

import org.apache.spark.sql.protobuf.functions._
import scala.collection.JavaConverters._

val schemaRegistryOptions = Map(
    "schema.registry.subject" -> "app-events-value",
    "schema.registry.address" -> "https://schema-registry:8081/"
)

// Convert binary Protobuf to SQL struct with from_protobuf():
val protoEventsDF = inputDF
    .select(
        from_protobuf($"proto_bytes", options = schemaRegistryOptions.asJava)
            .as("proto_event")
    )

// Convert SQL struct to binary Protobuf with to_protobuf():
val protobufBinaryDF = protoEventsDF
    .selectExpr("struct(name, id, context) as event")
    .select(
        to_protobuf($"event", options = schemaRegistryOptions.asJava)
            .as("proto_bytes")
    )

Autenticar em um registro de esquema confluente externo

Para autenticar em um Registro de Esquema Confluent externo, atualize as opções do Registro de esquema para incluir credenciais de autenticação e chaves de API.

Python

schema_registry_options = {
    "schema.registry.subject" : "app-events-value",
    "schema.registry.address" : "https://remote-schema-registry-endpoint",
    "confluent.schema.registry.basic.auth.credentials.source" : "USER_INFO",
    "confluent.schema.registry.basic.auth.user.info" : "confluentApiKey:confluentApiSecret"
  }

Scala

val schemaRegistryOptions = Map(
      "schema.registry.subject" -> "app-events-value",
      "schema.registry.address" -> "https://remote-schema-registry-endpoint",
      "confluent.schema.registry.basic.auth.credentials.source" -> "USER_INFO",
      "confluent.schema.registry.basic.auth.user.info" -> "confluentApiKey:confluentApiSecret"
)

Use arquivos truststore e keystore em volumes do Unity Catalog

No Databricks Runtime 14.3 LTS e superior, você pode usar arquivos truststore e keystore nos volumes do Unity Catalog para autenticar em um Registro de Esquema Confluent. Atualize as opções do Registro do esquema de acordo com o exemplo a seguir:

Python

schema_registry_options = {
    "schema.registry.subject" : "app-events-value",
    "schema.registry.address" : "https://remote-schema-registry-endpoint",
    "confluent.schema.registry.ssl.truststore.location" : "/Volumes/<catalog_name>/<schema_name>/<volume_name>/kafka.client.truststore.jks",
    "confluent.schema.registry.ssl.truststore.password" : "<password>",
    "confluent.schema.registry.ssl.keystore.location" : "/Volumes/<catalog_name>/<schema_name>/<volume_name>/kafka.client.keystore.jks",
    "confluent.schema.registry.ssl.keystore.password" : "<password>",
    "confluent.schema.registry.ssl.key.password" : "<password>"
  }

Scala

val schemaRegistryOptions = Map(
      "schema.registry.subject" -> "app-events-value",
      "schema.registry.address" -> "https://remote-schema-registry-endpoint",
      "confluent.schema.registry.ssl.truststore.location" -> "/Volumes/<catalog_name>/<schema_name>/<volume_name>/kafka.client.truststore.jks",
      "confluent.schema.registry.ssl.truststore.password" -> "<password>",
      "confluent.schema.registry.ssl.keystore.location" -> "/Volumes/<catalog_name>/<schema_name>/<volume_name>/kafka.client.keystore.jks",
      "confluent.schema.registry.ssl.keystore.password" -> "<password>",
      "confluent.schema.registry.ssl.key.password" -> "<password>"
)

Use o Protobuf com um arquivo descritor

Você também pode fazer referência a um arquivo descritor protobuf que está disponível para seu cluster de computação. Certifique-se de que tem as permissões adequadas para ler o ficheiro, dependendo da sua localização.

Python

from pyspark.sql.protobuf.functions import to_protobuf, from_protobuf

descriptor_file = "/path/to/proto_descriptor.desc"

proto_events_df = (
    input_df.select(
      from_protobuf(input_df.value, "BasicMessage", descFilePath=descriptor_file).alias("proto")
    )
)

proto_binary_df = (
  proto_events_df
    .select(
      to_protobuf(proto_events_df.proto, "BasicMessage", descriptor_file).alias("bytes")
    )
)

Scala

import org.apache.spark.sql.protobuf.functions._

val descriptorFile = "/path/to/proto_descriptor.desc"

val protoEventsDF = inputDF
  .select(
    from_protobuf($"value", "BasicMessage", descFilePath=descriptorFile).as("proto")
  )

val protoBytesDF = protoEventsDF
  .select(
    to_protobuf($"proto", "BasicMessage", descriptorFile).as("bytes")
  )

Opções suportadas nas funções do Protobuf

As seguintes opções são suportadas nas funções do Protobuf.

• mode: Determina como os erros durante a desserialização de registros Protobuf são tratados. Os erros podem ser causados por vários tipos de registros malformados, incluindo uma incompatibilidade entre o esquema real do registro e o esquema esperado fornecido no from_protobuf().
  • Valores:
    • FAILFAST(padrão): um erro é gerado quando um registro malformado é encontrado e a tarefa falha.
    • PERMISSIVE: Um NULL é retornado para registros malformados. Use esta opção com cuidado, pois pode resultar na queda de muitos registros. Isso é útil quando uma pequena fração dos registros na fonte está incorreta.
• recursive.fields.max.depth: Adiciona suporte para campos recursivos. Os esquemas do Spark SQL não oferecem suporte a campos recursivos. Quando esta opção não é especificada, os campos recursivos não são permitidos. A fim de suportar campos recursivos em Protobufs, eles precisam estar se expandindo para uma profundidade especificada.

  • Valores:

    • -1 (padrão): Campos recursivos não são permitidos.

    • 0: Os campos recursivos são descartados.

    • 1: Permite um único nível de recursão.

    • [2-10]: Especifique um limite para recursão múltipla, até 10 níveis.

      Definir um valor como maior que 0 permite campos recursivos expandindo os campos aninhados para a profundidade configurada. Valores maiores que 10 não são permitidos para evitar a criação inadvertida de esquemas muito grandes. Se uma mensagem Protobuf tiver profundidade além do limite configurado, a estrutura do Spark retornada será truncada após o limite de recursão.

  • Exemplo: Considere um Protobuf com o seguinte campo recursivo:

    message Person { string name = 1; Person friend = 2; }
    

    A seguir está listado o esquema final com valores diferentes para essa configuração:

    • Opção definida como 1: STRUCT<name: STRING>
    • Opção definida como 2: STRUCT<name STRING, friend: STRUCT<name: STRING>>
    • Opção definida como 3: STRUCT<name STRING, friend: STRUCT<name STRING, friend: STRUCT<name: STRING>>>
• convert.any.fields.to.json: Esta opção permite converter campos Protobuf Any para JSON. Este recurso deve ser ativado com cuidado. A conversão e o processamento JSON são ineficientes. Além disso, o campo de string JSON perde a segurança do esquema Protubuf, tornando o processamento downstream propenso a erros.

  • Valores:

    • Falso (padrão): No tempo de execução, esses campos curinga podem conter mensagens arbitrárias do Protobuf como dados binários. Por padrão, esses campos são tratados como uma mensagem Protobuf normal. Tem dois campos com esquema (STRUCT<type_url: STRING, value: BINARY>). Por padrão, o campo binário value não é interpretado de forma alguma. Mas os dados binários podem não ser convenientes na prática para trabalhar em alguns aplicativos.
    • True: Definir esse valor como True permite converter Any campos em cadeias de caracteres JSON em tempo de execução. Com essa opção, o binário é analisado e a mensagem Protobuf é desserializada em uma cadeia de caracteres JSON.

  • Exemplo: Considere dois tipos de Protobuf definidos da seguinte forma:

    message ProtoWithAny {
       string event_name = 1;
       google.protobuf.Any details = 2;
    }
    
    message Person {
       string name = 1;
       int32 id = 2;
    }
    

    Com essa opção habilitada, o esquema para from_protobuf("col", messageName ="ProtoWithAny") seria: STRUCT<event_name: STRING, details: STRING>.

    Em tempo de execução, se details o campo contiver Person a mensagem Protobuf, o valor retornado terá esta aparência: ('click', '{"@type":"type.googleapis.com/...ProtoWithAny","name":"Mario","id":100}').

  • Requisitos:

    • As definições para todos os possíveis tipos de Protobuf que são usados em Any campos devem estar disponíveis no arquivo descritor Protobuf passado para from_protobuf().
    • Se Any o Protobuf não for encontrado, isso resultará em um erro para esse registro.
    • Este recurso não é suportado atualmente com o registro de esquema.
• emit.default.values: Permite renderizar campos com valores zero ao desserializar o Protobuf para uma estrutura do Spark. Esta opção deve ser utilizada com moderação. Geralmente, não é aconselhável depender de diferenças semânticas tão finas.

  • Valores

    • Falso (padrão): Quando um campo está vazio no Protobuf serializado, o campo resultante na estrutura do Spark é, por padrão, nulo. É mais simples não habilitar essa opção e tratar null como o valor padrão.
    • True: Quando esta opção está ativada, esses campos são preenchidos com os valores padrão correspondentes.

  • Exemplo: Considere o seguinte Protobuf com o Protobuf construído como Person(age=0, middle_name=""):

    syntax = "proto3";
    
    message Person {
       string name = 1;
       int64 age = 2;
       optional string middle_name = 3;
       optional int64 salary = 4;
    }
    

    • Com essa opção definida como False, a estrutura do Spark após a chamada from_protobuf() seria toda nulls: {"name": null, "age": null, "middle_name": "", "salary": null}. Apesar de dois campos (age e middle_name) terem valores definidos, o Protobuf não os inclui no formato wire, uma vez que são valores padrão.
    • Com essa opção definida como True, a estrutura do Spark após a chamada from_protobuf() seria: {"name": "", "age": 0, "middle_name": "", "salary": null}. O salary campo permanece nulo uma vez que é explicitamente declarado optional e não está definido no registro de entrada.
• enums.as.ints: Quando ativado, os campos enum no Protobuf são renderizados como campos inteiros no Spark.

  • Valores

    • Falso (predefinição)
    • True: Quando ativado, os campos enum no Protobuf são renderizados como campos inteiros no Spark.

  • Exemplo: Considere o seguinte Protobuf:

    syntax = "proto3";
    
    message Person {
       enum Job {
         NONE = 0;
         ENGINEER = 1;
         DOCTOR = 2;
         NURSE = 3;
       }
       Job job = 1;
    }
    

    Dada uma mensagem Protobuf como Person(job = ENGINEER):

    • Com esta opção desativada, a estrutura do Spark correspondente seria {"job": "ENGINEER"}.
    • Com essa opção ativada, a estrutura do Spark correspondente seria {"job": 1}.

    Observe que o esquema para esses campos é diferente em cada caso (inteiro em vez de cadeia de caracteres padrão). Tal alteração pode afetar o esquema das tabelas a jusante.

Opções do Registro de Esquema

As seguintes opções de registro de esquema são relevantes ao usar o registro de esquema com funções Protubuf.

• schema.registry.subject
  • Necessário
  • Especifica o assunto para o esquema no Registro de Esquema, como "client-event"
• schema.registry.address
  • Necessário
  • URL para registro de esquema, como https://schema-registry.example.com:8081
• schema.registry.protobuf.name
  • Opcional
  • Padrão: <NONE>.
  • Uma entrada de registro de esquema para um assunto pode conter várias definições de Protobuf, assim como um único proto arquivo. Quando esta opção não é especificada, o primeiro Protobuf é usado para o esquema. Especifique o nome da mensagem Protobuf quando ela não for a primeira na entrada. Por exemplo, considere uma entrada com duas definições de Protobuf: "Pessoa" e "Local" nessa ordem. Se o fluxo corresponder a "Local" em vez de "Pessoa", defina esta opção como "Local" (ou seu nome completo, incluindo o pacote "com.example.protos.Location").
• schema.registry.schema.evolution.mode
  • Padrão: "reiniciar".
  • Modos suportados:
    • "Reiniciar"
    • "nenhuma"
  • Esta opção define o modo de evolução do esquema para from_protobuf(). No início de uma consulta, o Spark registra a id de esquema mais recente para determinado assunto. Isso determina o esquema para from_protobuf(). Um novo esquema pode ser publicado no registro do esquema após o início da consulta. Quando uma id de esquema mais recente é notada em um registro de entrada, isso indica uma alteração no esquema. Esta opção determina como essa alteração no esquema é tratada:
    • restart (padrão): aciona um UnknownFieldException quando uma id de esquema mais recente é notada. Isso encerra a consulta. O Databricks recomenda configurar trabalhos para reiniciar em caso de falha na consulta para pegar alterações de esquema.
    • none: As alterações de ID de esquema são ignoradas. Os registros com id de esquema mais recente são analisados com o mesmo esquema que foi observado no início da consulta. Espera-se que as definições mais recentes do Protobuf sejam compatíveis com versões anteriores e que novos campos sejam ignorados.
• confluent.schema.registry.<schema-registy-client-option>
  • Opcional
  • O Schema-registry se conecta ao Confluent schema-registry usando o cliente Confluent Schema Registry. Todas as opções de configuração suportadas pelo cliente podem ser especificadas com o prefixo "confluent.schema.registry". Por exemplo, as duas configurações a seguir fornecem credenciais de autenticação "USER_INFO":
    • "confluent.schema.registry.basic.auth.credentials.source": 'USER_INFO'
    • "confluent.schema.registry.basic.auth.user.info": "<KEY> : <SECRET>"