Configurar un flujo

Un objeto Stream representa un origen de datos de streaming externo, como Apache Kafka. Los flujos almacenan detalles de conexión, autenticación, esquemas y configuración de ingesta. Una vez creado un flujo, puede hacer referencia a él mediante definiciones de Feature View para crear funciones de streaming en tiempo real.

Los flujos tienen nombres de tres partes (catalog.schema.stream_name). El acceso a un flujo se rige por su tabla de ingestión asociada. Consulta Ingestión y relleno para más detalles.

Requisitos

Para ejecutar comandos de cuaderno: sin servidor o un clúster de proceso clásico que ejecuta Databricks Runtime 17.0 ML o superior.
Debe estar instalado el feature-engineering-client paquete de Python versión 0.18.0 o superior.

Conectarse a fuentes de streaming

Antes de definir las funciones de streaming, conecte y pruebe la conexión de una canalización de Lakeflow Streaming con su broker de Kafka. Feature Store depende de un SDP serverless, lo que significa que necesitarás un mecanismo para conectar tu computación clásica (broker o endpoint) con la computación serverless de Databrick. Esto se hace a través de productos como privatelink o permitiendo que tu computación clásica sea accesible desde internet público.

Crear un flujo

Usa create_stream() para crear un nuevo Stream. Un Stream requiere cuatro componentes de configuración:

Configuración de fuente: Especifica la plataforma de streaming y detalles específicos de la fuente, como la suscripción al tema de una fuente Kafka.
Configuración de conexión: especifica cómo conectarse y autenticarse en la plataforma de streaming, incluidos los servidores de arranque y las credenciales.
Configuración del esquema: define la estructura de las claves y los valores del mensaje.
Configuración de ingesta: especifica dónde y cómo se ingieren los datos de flujo. Consulta Ingestión y relleno para más detalles.

Para la configuración específica source_config de la fuente y la conexión, junto con un ejemplo completo create_stream() , véase Apache Kafka. Las opciones de esquema y de ingestión se comparten entre fuentes.

Apache Kafka

Para transmitir desde Apache Kafka, usa KafkaStreamConfig como configuración de origen y una conexión de Unity Catalog para autenticación. Consulte Streaming en proceso sin servidor y Conectarse a Apache Kafka para obtener información sobre la conectividad con Kafka.

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    KafkaStreamConfig,
    KafkaSubscriptionMode,
    StreamConnectionConfig,
    DirectSchemas,
    SchemaConfig,
    IngestionConfig,
    IngestionDestination,
    StreamBackfillSource,
)

client = FeatureEngineeringClient()

stream = client.create_stream(
    name="my_catalog.my_schema.my_stream",
    source_config=KafkaStreamConfig(
        subscription_mode=KafkaSubscriptionMode(subscribe="events-topic"),
    ),
    connection_config=StreamConnectionConfig(
        uc_connection_name="my-kafka-connection"
    ),
    schema_config=DirectSchemas(
        payload_schema=SchemaConfig(
            json_schema=(
                '{'
                '  "type": "object",'
                '  "properties": {'
                '    "transaction_id": {"type": "string"},'
                '    "user_id": {"type": "string"},'
                '    "amount": {"type": "number"},'
                '    "event_time": {"type": "string", "format": "date-time"}'
                '  }'
                '}'
            )
        ),
    ),
    ingestion_config=IngestionConfig(
        ingestion_destination=IngestionDestination(
            delta_table_name="my_catalog.my_schema.events_ingestion"
        ),
    ),
)

Modos de suscripción Kafka

El modo de suscripción especifica cómo Stream selecciona los topics de Kafka de los que consume. Se admiten tres modos:

Modo	Description	Ejemplo
`subscribe`	Lista separada por comas de nombres de temas	`KafkaSubscriptionMode(subscribe="topic1,topic2")`
`subscribe_pattern`	Coincidencia de nombres de tema con patrones de expresiones regulares en Java	`KafkaSubscriptionMode(subscribe_pattern="events-.*")`
`assign`	JSON que especifica asignaciones de particiones de tema	`KafkaSubscriptionMode(assign='{"my-topic": [0, 1, 2]}')`

Autenticación de Kafka

Conexión de Unity Catalog (recomendado)

Use una conexión de Catálogo de Unity para autenticarse en el clúster de Kafka. Este es el enfoque recomendado para la autenticación administrada. Para crear una conexión, consulte Creación de una conexión. El creador del flujo debe tener USE CONNECTION en la conexión. Cualquier usuario que cree funcionalidades a partir de transmisiones como origen también debe tener USE CONNECTION en la conexión.

connection_config = StreamConnectionConfig(
    uc_connection_name="my-kafka-connection"
)

La conexión soporta tanto autenticación IAM (credencial de servicio) como SASL.

IAM (credencial de servicio)

Autentica con una credencial de servicio de Unity Catalog, por ejemplo, para conectarte a Amazon MSK con IAM. Para crear una credencial de servicio, consulte Creación de credenciales de servicio. Establece el nombre de la credencial del servicio con la credential opción:

CREATE CONNECTION IF NOT EXISTS `my-kafka-connection`
TYPE KAFKA
OPTIONS (
    bootstrap_servers '<bootstrap_servers>',
    credential '<service_credential>'
)

Además de USE CONNECTION en la conexión, las identidades que utilizan la credencial de servicio necesitan ACCESS en ella. Conceda ACCESS en la credencial de servicio indicada al creador de la transmisión y a cualquier identidad que materialice características de la transmisión. Consulte Concesión de permisos para usar una credencial de servicio para acceder a un servicio en la nube externo.

SASL

La autenticación SASL utiliza un nombre de usuario y una contraseña. Establezca en sasl_mechanism una de las siguientes opciones:

PLAIN
SCRAM-SHA-256
SCRAM-SHA-512

Proporciona las credenciales con las opciones user y password. La conexión almacena estas credenciales de forma segura.

El siguiente ejemplo utiliza SASL/SCRAM. Para SASL/PLAIN, pon sasl_mechanism en PLAIN.

CREATE CONNECTION IF NOT EXISTS `my-kafka-connection`
TYPE KAFKA
OPTIONS (
    bootstrap_servers '<bootstrap_servers>',
    sasl_mechanism 'SCRAM-SHA-512',
    user '<username>',
    password '<password>'
)

MTLS directo

Para la autenticación mTLS directa, proporcione archivos keystore y truststore almacenados en un volumen de Unity Catalog, con contraseñas referenciadas mediante secret scopes de Databricks. Para más información sobre la autenticación SSL con Kafka, consulte Uso de SSL para conectarse Azure Databricks a Kafka.

from databricks.feature_engineering.entities import (
    DirectMtlsConfig,
    MtlsConfig,
    SecretScopeReference,
)

connection_config = DirectMtlsConfig(
    bootstrap_servers="broker1:9092,broker2:9092",
    mtls_config=MtlsConfig(
        keystore_location="/Volumes/my_catalog/my_schema/my_volume/keystore.jks",
        keystore_password_ref=SecretScopeReference(
            scope="my_scope", key="keystore_password"
        ),
        key_password_ref=SecretScopeReference(
            scope="my_scope", key="key_password"
        ),
        truststore_location="/Volumes/my_catalog/my_schema/my_volume/truststore.jks",
        truststore_password_ref=SecretScopeReference(
            scope="my_scope", key="truststore_password"
        ),
    ),
)

Configuración del esquema

Define la estructura de las claves y valores de los mensajes para que las definiciones de ingestión y características puedan leer campos individuales. Para los orígenes de Kafka, payload_schema corresponde al valor del mensaje de Kafka (en el value modelo clave-valor de Kafka) y key_schema corresponde a la clave de mensaje de Kafka. Se debe proporcionar al menos uno de payload_schema o key_schema .

Cada uno SchemaConfig acepta uno de tres formatos, coincidiendo con cómo la fuente serializa sus mensajes: json_schema, avro_schema, o proto_schema. Si no se proporciona ningún esquema para una clave o carga, se trata como una cadena simple.

Los ejemplos de código de esta sección utilizan esquemas declarados en línea con DirectSchemas, donde el esquema se proporciona como una cadena. Para gestionar esquemas utilizando un registro externo de esquemas, consulte Registro de esquemas para más detalles.

Esquema JSON

Proporciona una cadena de esquema JSON a json_schema.

schema_config = DirectSchemas(
    payload_schema=SchemaConfig(
        json_schema=(
            '{'
            '  "type": "object",'
            '  "properties": {'
            '    "user_id": {"type": "string"},'
            '    "amount": {"type": "number"},'
            '    "event_time": {"type": "string"}'
            '  }'
            '}'
        )
    ),
    key_schema=SchemaConfig(
        json_schema='{"type": "string"}'
    ),
)

Esquema de Avro

Proporciona una cadena de esquema Avro a avro_schema. Se admiten tipos lógicos Avro, incluyendo timestamp-millis, date, y decimal.

schema_config = DirectSchemas(
    payload_schema=SchemaConfig(
        avro_schema=(
            '{'
            '  "type": "record",'
            '  "name": "Event",'
            '  "fields": ['
            '    {"name": "user_id", "type": "string"},'
            '    {"name": "amount", "type": "double"},'
            '    {"name": "event_time",'
            '     "type": {"type": "long", "logicalType": "timestamp-millis"}}'
            '  ]'
            '}'
        )
    ),
)

Esquema Protobuf

Proporcione un ProtoSchemaSpec a proto_schema con el texto de origen de Búferes de protocolo.proto y el nombre del mensaje de la carga. Importa ProtoSchemaSpec desde databricks.feature_engineering.entities.

message_name debe ser el nombre del mensaje completamente cualificado, incluyendo el package declarado en el .proto texto (por ejemplo, com.example.Event, no Event). Se admiten tanto la sintaxis proto2 como la proto3.

google.protobuf.Timestamp y los tipos de envoltorios escalares (StringValue, Int32Value, y así sucesivamente) son soportados, y sus importaciones se resuelven automáticamente. Otros tipos conocidos, como Duration, Struct, y Any, son rechazados; codifica esos valores como un escalar o mensaje soportado en su lugar. Los tipos escalares fixed32 y fixed64, así como map con claves que no sean de cadena, tampoco son compatibles.

from databricks.feature_engineering.entities import ProtoSchemaSpec

schema_config = DirectSchemas(
    payload_schema=SchemaConfig(
        proto_schema=ProtoSchemaSpec(
            schema_text=(
                'syntax = "proto3";\n'
                'package com.example;\n'
                'import "google/protobuf/timestamp.proto";\n'
                'message Event {\n'
                '  string user_id = 1;\n'
                '  double amount = 2;\n'
                '  google.protobuf.Timestamp event_time = 3;\n'
                '}'
            ),
            message_name="com.example.Event",
        )
    ),
)

Decodificación de datos usando esquemas

Databricks decodifica cada mensaje con las funciones de from_jsonSpark , from_avro, y from_protobuf . Se aplican los siguientes comportamientos tanto si declaras el esquema en línea como si lo resuelves desde un registro de esquemas:

Registros malformados. La decodificación utiliza el modo PERMISSIVE, por lo que un registro que no coincide con su esquema da como resultado un valor nulo en lugar de provocar un error en el flujo.
Sindicatos Avro. Una unión de múltiples tipos de registro se decodifica como una estructura con un campo por cada tipo de registro, y cada campo recibe el nombre de su registro de Avro correspondiente.
Tipos Protobuf. Los enteros sin signo se decodifican como un tipo con signo de mayor tamaño (por ejemplo, uint32 a BIGINT y uint64 a DECIMAL(20,0)), los campos de enumeración se decodifican como su nombre en forma de cadena, y los tipos envoltorio escalares (por ejemplo, StringValue y Int32Value) se decodifican como una columna que admite valores NULL del tipo envuelto.

Registro de esquemas

Los registros de esquemas almacenan y versionan los esquemas que usan los productores y consumidores de streaming, haciendo cumplir las reglas de compatibilidad a medida que esos esquemas evolucionan. Cuando se configura un registro de esquema externo, Feature Store lee el esquema del registro y lo utiliza para decodificar el mensaje en streaming. No declaras el esquema en línea en el Stream cuando usas un registro de esquema.

El soporte para registros de esquemas tiene las siguientes limitaciones:

Compatible únicamente con flujos de Kafka.
Solo es compatible Confluent Schema Registry
Solo se soportan los formatos Avro y Protobuf . Para leer mensajes JSON, declara el esquema en línea en su lugar. Ver esquema JSON.
Cada Flujo está conectado exactamente a un sujeto Confluente para el valor del mensaje y uno para la clave del mensaje (si se proporciona). Los temas de flujo que contienen múltiples registros de esquema no son una configuración soportada. Si tu Stream se conecta a temas que contienen múltiples esquemas, los registros que no coinciden con el esquema del tema especificado se decodifican como nulos.

Conectarse a un registro de esquemas

Proporcione los detalles de la conexión del registro como opciones en la conexión del Catálogo Kafka Unity y almacene el secreto de la API del registro en un ámbito secreto de Databricks. La identidad de ejecución de transmisión debe tener permiso READ en el ámbito de secretos, porque la canalización de ingesta lee el secreto en tiempo de ejecución. Para saber cómo crear y configurar una conexión, véase Crear una conexión.

Añade las schema_registry_urlopciones , schema_registry_api_key, y schema_registry_api_secret a la conexión utilizada para la autenticación. El siguiente ejemplo crea una conexión Kafka que se autentica con el broker con una credencial de servicio de Unity Catalog y con el registro con una clave API:

CREATE CONNECTION IF NOT EXISTS `my-kafka-connection`
TYPE KAFKA
OPTIONS (
    bootstrap_servers '<bootstrap_servers>',
    credential '<service_credential>',
    schema_registry_url 'https://<registry-host>',
    schema_registry_api_key '<registry_api_key>',
    schema_registry_api_secret secret('<scope>', '<key>')
)

Establezca tanto la opción schema_registry_api_secret en la conexión de Kafka como la referencia al ámbito de secreto en el flujo para que apunten al mismo secreto.

Crea un flujo que utilice un registro de esquema

Pase a SchemaRegistryConfig como el schema_config. Referencia el secreto de la API del registro con api_secret_ref, e identifica el asunto y formato con payload_schema_locator para el valor del mensaje, o key_schema_locator para la clave del mensaje. Debe proporcionarse al menos un localizador.

Observa las diferencias aquí en comparación con los ejemplos directos de esquemas en la sección de configuración de esquemas . Al usar un registro de esquema, no se proporciona el esquema en línea en la transmisión a schema_config. En su lugar, especificas una SchemaRegistryConfig que identifica el esquema en el registro.

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    KafkaStreamConfig,
    KafkaSubscriptionMode,
    StreamConnectionConfig,
    SchemaRegistryConfig,
    SchemaLocator,
    SchemaLocatorConfluentSchema,
    SchemaLocatorFormat,
    SecretScopeReference,
    IngestionConfig,
    IngestionDestination,
)

client = FeatureEngineeringClient()

stream = client.create_stream(
    name="my_catalog.my_schema.my_stream",
    source_config=KafkaStreamConfig(
        subscription_mode=KafkaSubscriptionMode(subscribe="transactions"),
    ),
    connection_config=StreamConnectionConfig(
        uc_connection_name="my-kafka-connection"
    ),
    schema_config=SchemaRegistryConfig(
        api_secret_ref=SecretScopeReference(
            scope="my_scope", key="sr_api_secret"
        ),
        payload_schema_locator=SchemaLocator(
            confluent_schema=SchemaLocatorConfluentSchema(
                subject="transactions-value"
            ),
            format=SchemaLocatorFormat.FORMAT_AVRO,
        ),
    ),
    ingestion_config=IngestionConfig(
        ingestion_destination=IngestionDestination(
            delta_table_name="my_catalog.my_schema.transactions_ingestion"
        ),
    ),
)

Un sujeto Confluente es el ámbito denominado bajo el cual se registra el historial de versiones de un esquema y se hace cumplir la compatibilidad. Establezca subject el nombre del ámbito correspondiente, que de forma común se determina a partir de la estrategia de nombres de sujeto:

TopicNameStrategy (por defecto, deriva el tema del nombre del tema): <topic>-value para el valor y <topic>-key para la clave. Por ejemplo, el esquema de valores del tema transactions utiliza el sujeto transactions-value.
RecordNameStrategy (deriva el sujeto del nombre del registro del esquema, independientemente del tema): el nombre de registro completamente cualificado, como com.example.Payment. Este es el espacio de nombres y el nombre del registro para Avro, o el paquete y el nombre del mensaje para Protobuf.
TopicRecordNameStrategy (combina los nombres de tema y registro): <topic>-<fully-qualified-record-name>, como transactions-com.example.Payment.

format es obligatorio. Configúralo para SchemaLocatorFormat.FORMAT_AVRO o SchemaLocatorFormat.FORMAT_PROTOBUF para que coincida con cómo se serializa el tema.

Evolución del esquema

La canalización de ingesta resuelve el esquema actual del sujeto cuando se inicia. Cuando registra una nueva versión de esquema compatible con versiones anteriores en el sujeto en el registro de esquemas, la canalización en ejecución sigue usando la versión con la que empezó.

En el caso de los Streams respaldados por un registro de esquemas, el proceso de ingesta se reinicia automáticamente cada dos horas aproximadamente. En cada reinicio, recoge la última versión del esquema del tema, y los campos nuevos o modificados aparecen en la tabla de ingesta.

Los flujos que usan esquemas directos en lugar de un registro de esquemas evolucionan su esquema con update_stream. Consulta Actualizar un stream.

Para ver cómo la pipeline gestiona los registros que no coinciden con el esquema que está usando actualmente, véase Decodificación de datos usando esquemas.

Filtrar registros por tipo

Un flujo decodifica cada registro usando un único esquema de clave y valor (si se proporciona), ya sea que los especifiques directamente o utilices un registro de esquemas. Dado que un tema puede contener más de un tipo de registro y que los Streams pueden suscribirse a varios temas, utiliza record_type_filter para seleccionar qué registros del tema pertenecen a este Stream.

Proporciona una expresión SQL que haga referencia a los campos descodificados con notación de puntos, por ejemplo value.event_type = 'transaction'. Los registros que no coincidan con el filtro se ignoran. No se escriben en la tabla de ingesta y no se utilizan en la materialización. Para crear un flujo para otros tipos de registro, crea un flujo separado con otro record_type_filter.

stream = client.create_stream(
    name="my_catalog.my_schema.my_stream",
    # ...source, connection, schema, and ingestion config...
    record_type_filter="value.event_type = 'transaction'",
)

Incluso sin record_type_filter, la decodificación nunca provoca un error en el Stream. Un registro que no se ajusta al esquema configurado se decodifica de forma permisiva. Los registros se decodifican de una de las siguientes maneras:

En una fila con valores NULL para los campos que el esquema espera pero que el registro omite (JSON, Avro y Protobuf).
En una fila que contiene valores que pertenecen a un tipo de registro diferente (solo Avro y Protobuf).

Para identificar si una fila en la tabla de ingestión pertenece al tipo de registro esperado, utilice una de las siguientes comprobaciones:

Comprobar que un campo sea igual a un valor esperado, por ejemplo value.event_type = 'transaction' (preferible para Avro y Protobuf).
Comprobar que un campo no sea NULL, por ejemplo value.activity_id IS NOT NULL.

Se recomienda utilizar record_type_filter con Streams independientes cuando los esquemas difieran sustancialmente entre los tipos de registro del tema o cuando se desee controlar el acceso a cada tipo de registro de forma independiente. Para controlar los costes, Databricks recomienda mantener un número reducido de Streams, ya que cada Stream cuenta con un canal de ingesta y una tabla de ingesta independientes. Además, cada Stream utiliza recursos de proceso independientes en el momento de la materialización. Se pueden utilizar filtros específicos de cada característica para la materialización.

record_type_filter es diferente del filter_condition de una característica. record_type_filter se configura en el Stream y controla qué registros se ingieren y están disponibles para todas las características que utilizan el Stream como fuente, mientras que filter_condition se configura en una característica individual y filtra las filas antes de la agregación. Consulta Condiciones de filtro en fuentes de streaming para más detalles sobre filter_condition.

Ingestión y relleno

El ingestion_config parámetro configura cómo se capturan y almacenan los datos de flujo para el entrenamiento y el servicio.

El acceso a un Stream se rige por la tabla de ingestión:

SELECT en la tabla de ingesta otorga acceso de lectura al Stream.
MANAGE en la tabla de ingesta concede acceso para eliminar.

Para obtener más información sobre los privilegios de tabla, consulte Table y la referencia de privilegios de Unity Catalog.

Canalización de ingesta

Cuando se crea un flujo, Databricks inicia una tubería de ingestión gestionada que lee continuamente los mensajes del flujo fuente y los escribe en una tabla Delta (la tabla de ingestión). La canalización se inicia en la posición más reciente del origen y se ejecuta de forma continua, capturando solo los mensajes nuevos que llegan después de que se crea el flujo. Esta tabla de ingesta se usa para el entrenamiento con características de streaming. Cuando se elimina un flujo, también se eliminan la canalización de ingesta y la tabla de ingesta.

Destino de la ingesta

ingestion_destination especifica el nombre de la tabla Delta de tres partes donde se escriben los datos del flujo.

ingestion_config = IngestionConfig(
    ingestion_destination=IngestionDestination(
        delta_table_name="my_catalog.my_schema.events_ingestion"
    ),
)

Esquema de tabla de ingesta

La tabla de ingestión contiene los datos del mensaje junto con las columnas de metadatos. Las columnas comunes se encuentran en cada fuente; las columnas kafka_* se encuentran solo en un flujo de Kafka.

Columna	Tipo	Source	Description
`key`	Varía (según `key_schema`)	Common	La clave del mensaje, estructurada según el esquema que proporcionaste.
`value`	Varía (según `payload_schema`)	Common	El valor del mensaje (payload), estructurado según el esquema que proporcionaste.
`stream_record_timestamp`	`TIMESTAMP`	Common	Marca de tiempo del registro. Para los datos rellenados hacia adelante, esta es la marca de tiempo de ingesta del origen. En el caso de los datos de relleno, los proporciona el cliente.
`record_source`	`STRING`	Common	Ya sea `"stream"` (relleno directo desde la transmisión en directo) o `"backfill"` (desde la fuente de relleno).
`kafka_topic`	`STRING`	Kafka	El tema kafka del que se consumió el registro.
`kafka_partition`	`INT`	Kafka	Partición de Kafka desde la que se consumió el registro.
`kafka_offset`	`LONG`	Kafka	El desfase de Kafka del registro dentro de su partición.

Origen de reposición

Dado que la canalización de llenado comienza a partir de la última posición del origen, no recoge los mensajes que existían antes de que se creara la transmisión. Para proporcionar cobertura histórica de datos para el entrenamiento, configure un origen de reposición opcional.

Cuando se configura un origen de reposición, Databricks ejecuta un trabajo único de MERGE INTO que copia las filas de reposición en la tabla de ingesta con record_source="backfill". MERGE solo se ejecuta después de que el comprobador de reposición confirme que la fuente de backfill y el flujo de relleno hacia delante tienen marcas de tiempo superpuestas (consulte Superposición entre los datos de backfill y el flujo en vivo). Si no se cumple la condición de superposición en un plazo de 2 días, MERGE se ejecuta de todos modos para evitar el bloqueo indefinidamente.

La tabla de reposición debe incluir una stream_record_timestamp columna de tipo TIMESTAMP en la zona horaria UTC. Las demás columnas de metadatos se transfieren si están presentes en la fuente de backfill o, en caso contrario, se establecen en NULL. Para Kafka, estos son kafka_topic, kafka_partition, y kafka_offset.

from databricks.feature_engineering.entities import StreamBackfillSource

ingestion_config = IngestionConfig(
    ingestion_destination=IngestionDestination(
        delta_table_name="my_catalog.my_schema.events_ingestion"
    ),
    backfill_source=StreamBackfillSource(
        delta_table_name="my_catalog.my_schema.historical_events"
    ),
)

Superposición entre los datos de reposición y streaming en vivo

Antes de ejecutar una combinación entre reposición y la tabla de ingesta, una comprobación de superposición compara las marcas de tiempo en las dos tablas:

Número máximo de reposición: el máximo stream_record_timestamp en el origen de reposición.
Mín. de ingesta: el mínimo stream_record_timestamp de filas (record_source="stream") en la tabla de ingesta.

La operación MERGE se lleva a cabo cuando la marca de tiempo más reciente de la carga de reposición supera en al menos 1 hora la marca de tiempo más antigua de la tabla de ingesta. Esta superposición garantiza que no haya huecos en la tabla de ingesta. Si no se cumple la condición de superposición en un plazo de 2 días, MERGE se ejecuta de todos modos para evitar el bloqueo indefinidamente.

Como la tubería de ingestión comienza desde la posición más reciente en la fuente, solo captura los mensajes que llegan después de que se crea el flujo. El origen de reposición debe contener datos que se extienden al intervalo de tiempo de ingesta, no solo hasta el tiempo de creación de la transmisión.

Por ejemplo, si crea una transmisión a las 3:00 p. m., la pipeline de reposición comienza a leer mensajes a partir de las 3:00 p. m. El origen de reposición debe incluir datos con marcas temporales hasta al menos las 4:00 p. m. (1 hora después del inicio de la reposición) para superar la comprobación de solapamiento. Esto significa que debe actualizar la tabla de reposición después de las 4:00 p. m. para asegurarse de que la tabla de ingesta no tiene huecos.

Desduplicación

Utilice deduplication_columns para especificar rutas de columnas a fin de identificar filas duplicadas durante la ingesta de datos entre los flujos de relleno retroactivo y de relleno progresivo. Use la notación de puntos para los campos anidados (por ejemplo, "value.user_id").

Elija columnas de desduplicación en función de los datos:

Si cada registro de la secuencia contiene un identificador único (por ejemplo, value.transaction_id), use esa columna para la desduplicación.
Si el origen de reposición incluye kafka_partition y kafka_offset columnas, úselos para identificar de forma única cada registro.
Si no se especifica ninguna columna de desduplicación, la clave de desduplicación predeterminada es la combinación completa de key, valuey stream_record_timestamp. Esto no se recomienda, ya que esta coincidencia estricta de criterios puede provocar fácilmente duplicados.

ingestion_config = IngestionConfig(
    ingestion_destination=IngestionDestination(
        delta_table_name="my_catalog.my_schema.events_ingestion"
    ),
    deduplication_columns=["value.transaction_id"],
)

Atribución de costos

Configura tags y budget_policy_id en el IngestionConfig para atribuir el coste de la ingesta administrada del Stream. Azure Databricks los aplica a la canalización de ingesta de Lakeflow y a sus trabajos de relleno hacia delante y hacia atrás cuando se crea el Stream.

Por ejemplo, los límites de etiquetas y cómo consultar el gasto atribuido, véase Costes de atributos con etiquetas y políticas de uso serverless.

Excluir columnas de un flujo

Utiliza excluded_columns para eliminar columnas específicas de un Stream que no desees ingestar. Una columna excluida no se escribe en la tabla de ingesta y no puede ser referenciada por una característica ni utilizada en el entrenamiento.

Especifica cada columna utilizando la notación de puntos en la clave o el valor del mensaje, como value.user.email o key.account_id. Estas columnas se omiten de los elementos key y value decodificados durante la ingesta, el relleno y la materialización. Si una ruta apunta a una estructura, también se eliminan todos sus campos anidados (por ejemplo, al eliminar value.address también se eliminan value.address.city y value.address.zip).

stream = client.create_stream(
    name="my_catalog.my_schema.my_stream",
    # ...source, connection, schema, and ingestion config...
    excluded_columns=["value.user.email", "value.user.ssn"],
)

Al usar esquemas directos, la columna excluida debe existir ya en el esquema clave o de valores, o create_stream falla. Cuando se utiliza un registro de esquemas, puedes excluir una columna antes de que exista. Una columna excluida tampoco puede ser una columna de deduplicación, ya que las columnas de deduplicación son necesarias para identificar filas duplicadas. Cualquier característica que haga referencia a una columna excluida (por ejemplo, como entidad, serie temporal o entrada) no se creará.

Se pueden modificar las columnas excluidas de un Stream tras su creación con update_stream, tanto en Streams con esquema directo como en aquellos respaldados por un registro de esquemas. Consulte Actualizar un Stream para obtener más detalles.

Gestionar flujos

Obtener una transmisión

stream = client.get_stream(name="my_catalog.my_schema.my_stream")

Listar flujos

streams = client.list_streams(
    catalog_name="my_catalog",
    schema_name="my_schema",
    max_results=50,
    include_schemas=False,
)

Establezca include_schemas=True para incluir todos los detalles del esquema. Los esquemas pueden ser grandes y esto puede dar lugar a una operación de larga duración. Para recuperar esquemas individualmente, use get_stream.

Actualizar un flujo

Utilice update_stream para modificar un Stream tras su creación. Introduce schema_config para actualizar un esquema directo, excluded_columns para cambiar las columnas que se eliminan, o ambos. No se admite la actualización de otros campos. En su lugar, crea un nuevo Stream.

Actualizar un flujo reinicia su pipeline de ingesta para que el cambio tenga efecto. La ingesta suele reanudarse en unos minutos.

Actualizar un esquema directo

Para un Stream que utiliza esquemas directos, pasa un DirectSchemas a schema_config. Establece payload_schema, key_schema o ambos. El lado que no se establezca permanecerá sin cambios. Los flujos respaldados por el registro de esquemas rechazan una actualización de schema_config y deben modificarse a través del registro.

from databricks.feature_engineering.entities import DirectSchemas, SchemaConfig

stream = client.update_stream(
    name="my_catalog.my_schema.my_stream",
    schema_config=DirectSchemas(
        payload_schema=SchemaConfig(
            json_schema=(
                '{'
                '  "type": "object",'
                '  "properties": {'
                '    "user_id": {"type": "string"},'
                '    "amount": {"type": "number"},'
                '    "event_time": {"type": "string"},'
                '    "channel": {"type": "string"}'
                '  }'
                '}'
            )
        ),
    ),
)

Las actualizaciones del esquema deben ser retrocompatibles para que la canalización de ingesta en ejecución pueda seguir descodificando los registros existentes y escribiendo en la tabla de ingesta. Se rechazan cualquier otro cambio.

Lo que está permitido depende del formato:

JSON y Protobuf: añadir campos opcionales, eliminar campos y ampliar el tipo de un campo (por ejemplo, de int a bigint). Protobuf también permite reordenar los campos.
Avro: solo permite ampliar de int a long y eliminar un campo final cuyos bytes no sean leídos por ningún campo posterior. Para evolucionar un esquema Avro de forma más libre, utiliza un flujo respaldado por un registro de esquemas.

Al añadir campos, aumentan las estructuras key y value descodificadas de la tabla de ingesta. Las filas escritas antes de la actualización conservan su forma original, y los campos añadidos se leen como NULL para esas filas anteriores. Las eliminaciones y los cambios de tipo solo surten efecto para los registros ingestados después de la actualización.

Cambiar las columnas excluidas

Pasa el nuevo conjunto completo de rutas de columnas a excluded_columns, lo que sustituye al conjunto existente. Pasa una lista vacía ([]) para borrar todas las exclusiones. Para obtener más detalles sobre este comportamiento, consulta Excluir columnas de un Stream.

stream = client.update_stream(
    name="my_catalog.my_schema.my_stream",
    excluded_columns=["value.user.email", "value.user.ssn"],
)

El cambio de las columnas excluidas solo tiene efecto prospectivo. Las columnas recién excluidas dejan de escribirse (aparecen en NULL) y las recién incluidas comienzan a rellenarse a partir de ese momento, mientras que las filas escritas anteriormente se mantienen tal cual. Para evitar que una nueva columna se incorpore en algún momento:

Registro de esquemas: añade primero la columna a excluded_columns y espera a que se reinicie el canal de ingesta; a continuación, registra la nueva versión del esquema en el registro.
Esquemas directos: añade la columna a schema_config y a excluded_columns en la misma llamada a update_stream.

Eliminar un flujo

Al eliminar un flujo, también se eliminan su canalización de ingesta y su tabla de ingesta.

Warning

Los modelos o características que hacen referencia a la secuencia eliminada ya no tendrán acceso a los datos de la secuencia subyacente. Cree una copia de la tabla de ingesta antes de eliminarla si necesita estos datos, pero ya no la transmisión.

client.delete_stream(name="my_catalog.my_schema.my_stream")

Cuaderno de ejemplo

Para ver un ejemplo de principio a fin que crea una transmisión, define las características de transmisión y despliega en un punto de conexión de servicio, consulte el siguiente cuaderno:

Cuaderno de inicio rápido para vistas de atributos de streaming

Obtención del cuaderno

Comentarios

¿Le resultó útil esta página?

Última actualización el 2026-10-12