Ingestion de journaux en temps réel dans un lac de données - Realtime Compute for Apache Flink

Cette rubrique présente les meilleures pratiques pour écrire des données de journal dans Alibaba Cloud Data Lake Formation (DLF) à l'aide d'un job YAML d'ingestion de données Flink CDC.

Ingestion de journaux Kafka dans un lac de données

Utilisez l'ingestion de données Flink CDC et un simple job YAML pour ingérer rapidement des données de journal dans un lac de données en temps réel. Le système infère automatiquement le schéma et prend en charge son évolution.

Supposons que le topic inventory dans Apache Kafka stocke les données d'une table de journal au format JSON. L'exemple de job suivant synchronise ces données vers la table de destination correspondante dans Data Lake Formation (DLF) :

source:
  type: kafka
  name: Kafka Source
  # The Kafka broker addresses.
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  # The topic to consume.
  topic: inventory
  # Specifies that consumption starts from the earliest offset.
  scan.startup.mode: earliest-offset
  # The format of the Kafka message value.
  value.format: json
  # Sets the schema inference strategy to continuous. This strategy detects the schema of each message and synchronizes schema changes.
  schema.inference.strategy: continuous
  # (Optional) Recursively flattens nested columns in JSON data.
  json.infer-schema.flatten-nested-columns.enable: true
  # (Optional) Skips the first 100 parsing exceptions. If the number of exceptions exceeds 100, the job fails.
  ingestion.ignore-errors: true
  ingestion.error-tolerance.max-count: 100

sink:
  type: paimon
  # The metastore type. The value is fixed to rest.
  catalog.properties.metastore: rest
  # The token provider. The value is fixed to dlf.
  catalog.properties.token.provider: dlf
  # The URI to access the DLF Rest Catalog Server. The format is http://[region-id]-vpc.dlf.aliyuncs.com. For example, http://cn-hangzhou-vpc.dlf.aliyuncs.com.
  catalog.properties.uri: dlf_uri
  # The name of the DLF catalog.
  catalog.properties.warehouse: your_warehouse
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

# Adds primary key information to the table. 
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id
    
# Writes all data from the inventory topic to the test_database.inventory table.
route:
  - source-table: inventory
    sink-table: test_database.inventory
    
pipeline:
  # (Optional) Records dirty data that causes processing exceptions in the logs.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Supposons que le topic inventory dans Apache Kafka stocke les données de plusieurs tables de journal au format JSON. Les champs databaseName et tableName dans la charge utile JSON fournissent les noms de base de données et de table. L'exemple de job suivant synchronise les données de ces tables vers leurs tables de destination respectives dans DLF :

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: json
  # (Optional) Recursively flattens nested columns in JSON data.
  json.infer-schema.flatten-nested-columns.enable: true
  # Uses the value of the databaseName field as the database name and the value of the tableName field as the table name.
  json.decode.parser-table-id.fields: databaseName,tableName
  # (Optional) Skips the first 100 parsing exceptions. If the number of exceptions exceeds 100, the job fails.
  ingestion.ignore-errors: true
  ingestion.error-tolerance.max-count: 100

sink:
  type: paimon
  # The metastore type. The value is fixed to rest.
  catalog.properties.metastore: rest
  # The token provider. The value is fixed to dlf.
  catalog.properties.token.provider: dlf
  # The URI to access the DLF Rest Catalog Server. The format is http://[region-id]-vpc.dlf.aliyuncs.com. For example, http://cn-hangzhou-vpc.dlf.aliyuncs.com.
  catalog.properties.uri: dlf_uri
  # The name of the DLF catalog.
  catalog.properties.warehouse: your_warehouse
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

# Adds primary key information to the table. 
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id
    
# Writes data from ods.inventory, ods.customer, and ods.user to the test_database.inventory, test_database.customer, and test_database.user tables, respectively.
route:
  - source-table: ods.inventory
    sink-table: test_database.inventory
  - source-table: ods.customer
    sink-table: test_database.customer
  - source-table: ods.user
    sink-table: test_database.user   
    
pipeline:
  # (Optional) Records dirty data that causes processing exceptions in the logs.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Pour plus de détails sur les politiques d'inférence et d'évolution de schéma pour les tables sources Kafka au format JSON, consultez Politiques d'analyse et d'évolution de schéma.

Cas d'utilisation

Les sections suivantes décrivent les configurations de job pour des cas d'utilisation courants. Pour des configurations plus détaillées, consultez Connecteur Kafka d'ingestion de données.

Résoudre les conflits de noms de champs

Un message Apache Kafka se compose d'une clé (Key) et d'une valeur (Value). Vous pouvez définir le format de chaque partie en configurant key.format et value.format. Le schéma final combine tous les champs des deux parties.

En cas de conflit de noms de champs entre les parties Key et Value, utilisez key.fields-prefix et value.fields-prefix pour ajouter des préfixes aux noms de champs.

Par exemple, si la partie Key d'un message Kafka contient les champs id et name, et que la partie Value contient les champs id et price, la configuration de job suivante produit un schéma avec les champs key_id, key_name, val_id et val_price.

source:
  type: kafka
  properties.bootstrap.servers: localhost:9092
  topic: test_topic
  properties.group.id: test_group
  scan.startup.mode: earliest-offset
  value.format: json
  key.format: json
  # Adds a prefix to field names in the Key.
  key.fields-prefix: key_
  # Adds a prefix to field names in the Value.
  value.fields-prefix: val_
  # Sets the schema inference strategy to continuous. This strategy detects the schema of each message and synchronizes schema changes.
  schema.inference.strategy: continuous
  # Ignores errors during data parsing.
  ingestion.ignore-errors: true
  # If 100 data parsing errors occur, the job fails.
  ingestion.error-tolerance.max-count: 100

# Writes all data from the test_topic topic to the test_database.test_topic table.
route:
  - source-table: test_topic
    sink-table: test_database.test_topic

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
pipeline:
  # Enables dirty data collection. Dirty data is written to a log file.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Lire les métadonnées

Utilisez la configuration metadata.list pour lire et transmettre des métadonnées supplémentaires du message Kafka. Les colonnes de métadonnées ajoutées dans metadata.list peuvent être utilisées directement dans le module transform. Pour obtenir la liste des métadonnées prises en charge, consultez Colonnes de métadonnées disponibles.

La configuration suivante ajoute les métadonnées partition et offset aux données. Elle utilise ensuite ces métadonnées dans le module transform pour filtrer les données dont la partition est supérieure à 1 et l'offset supérieur à 100.

source:
  type: kafka
  name: Kafka source
  properties.bootstrap.servers: localhost:9092
  topic: test_topic
  properties.group.id: test_group
  scan.startup.mode: earliest-offset
  value.format: json
  # Adds metadata columns.
  metadata.list: partition,offset
  # Sets the schema inference strategy to continuous. This strategy detects the schema of each message and synchronizes schema changes.
  schema.inference.strategy: continuous
  # Ignores errors during data parsing.
  ingestion.ignore-errors: true
  # If 100 data parsing errors occur, the job fails.
  ingestion.error-tolerance.max-count: 100
  
transform:
  - source-table: \.*.\.*
    filter: '`partition` > 1 and `offset` > 100'
    
# Writes all data from the test_topic topic to the test_database.test_topic table.
route:
  - source-table: test_topic
    sink-table: test_database.test_topic

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
pipeline:
  # Enables dirty data collection. Dirty data is written to a log file.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Gérer les données incorrectes

Les données de journal peuvent contenir des données incorrectes mal formatées, ce qui peut entraîner l'échec et le redémarrage répété du job. L'ingestion de données Flink CDC permet d'ignorer les erreurs d'analyse et de collecter les données dont l'analyse a échoué. Pour plus d'informations, consultez Collecte de données incorrectes.

Le job suivant écrit dans un fichier journal les données incorrectes dont l'analyse a échoué. Le job échoue si plus de 100 erreurs d'analyse se produisent.

source:
  type: kafka
  name: Kafka source
  properties.bootstrap.servers: localhost:9092
  topic: test_topic
  properties.group.id: test_group
  scan.startup.mode: earliest-offset
  value.format: json
  # Sets the schema inference strategy to continuous. This strategy detects the schema of each message and synchronizes schema changes.
  schema.inference.strategy: continuous
  # Ignores errors during data parsing.
  ingestion.ignore-errors: true
  # If 100 data parsing errors occur, the job fails.
  ingestion.error-tolerance.max-count: 100
  
# Writes all data from the test_topic topic to the test_database.test_topic table.
route:
  - source-table: test_topic
    sink-table: test_database.test_topic

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
pipeline:
  # Enables dirty data collection. Dirty data is written to a log file.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Si vous n'avez pas besoin de collecter les données incorrectes et souhaitez éviter les échecs de job dus à ces données, utilisez json.ignore-parse-errors, debezium-json.ignore-parse-errors ou canal-json.ignore-parse-errors pour ignorer directement les erreurs d'analyse.

Spécifier un TableID

Les formats Debezium JSON et Canal JSON sont fixes et utilisent des champs spécifiques pour stocker le TableID. Pour les données JSON génériques, qui n'ont pas de format fixe, le nom du topic sert par défaut de TableID. Pour utiliser des champs spécifiques des données comme TableID, configurez json.decode.parser-table-id.fields. Par exemple, étant donné les données JSON {"col0":"a", "col1":"b", "col2":"c"}, différentes configurations génèrent les TableIDs suivants :

Configuration

TableID

col0

col0,col1

a.b

col0,col1,col2

a.b.c

La configuration de job suivante concatène les colonnes db et table des données pour créer le TableID.

source:
  type: kafka
  name: Kafka source
  properties.bootstrap.servers: localhost:9092
  topic: test_topic
  properties.group.id: test_group
  scan.startup.mode: earliest-offset
  value.format: json
  # Uses the db and table fields from the data as the TableID.
  json.decode.parser-table-id.fields: db,table
  # Sets the schema inference strategy to continuous. This strategy detects the schema of each message and synchronizes schema changes.
  schema.inference.strategy: continuous
  # Ignores errors during data parsing.
  ingestion.ignore-errors: true
  # If 100 data parsing errors occur, the job fails.
  ingestion.error-tolerance.max-count: 100

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
pipeline:
  # Enables dirty data collection. Dirty data is written to a log file.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Inférer les types de données

L'analyse des données permet d'inférer les types de données des journaux. Pour des informations détaillées sur l'inférence de type, consultez Politiques d'analyse et d'évolution de schéma. Les sections suivantes expliquent comment ajuster les configurations pour contrôler l'analyse des types dans des scénarios courants.

(Général) Définir tous les types de champs sur String

Si le traitement et le stockage en aval ne nécessitent pas de types de champs spécifiques, activez json.infer-schema.primitive-as-string, debezium-json.infer-schema.primitive-as-string ou canal-json.infer-schema.primitive-as-string. Cela ignore l'inférence de type et définit tous les types de champs sur String.

source:
  type: kafka
  name: Kafka source
  properties.bootstrap.servers: localhost:9092
  topic: test_topic
  properties.group.id: test_group
  scan.startup.mode: earliest-offset
  value.format: json
  # Sets the type for all fields in the JSON data to String.
  json.infer-schema.primitive-as-string: true
  # Sets the schema inference strategy to continuous. This strategy detects the schema of each message and synchronizes schema changes.
  schema.inference.strategy: continuous
  # Ignores errors during data parsing.
  ingestion.ignore-errors: true
  # If 100 data parsing errors occur, the job fails.
  ingestion.error-tolerance.max-count: 100

# Writes all data from the test_topic topic to the test_database.test_topic table.
route:
  - source-table: test_topic
    sink-table: test_database.test_topic

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
pipeline:
  # Enables dirty data collection. Dirty data is written to a log file.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

(Général) Spécifier un schéma initial

Dans certains scénarios, vous devrez peut-être spécifier un schéma initial, par exemple lors de l'écriture de données Kafka dans une table existante en aval. Spécifiez un schéma initial en ajoutant le paramètre scan.value.initial-schemas.ddls.

source:
  type: kafka
  name: Kafka source
  properties.bootstrap.servers: localhost:9092
  topic: test_topic
  properties.group.id: test_group
  scan.startup.mode: earliest-offset
  value.format: json
  # Uses the db and table fields from the data as the TableID.
  json.decode.parser-table-id.fields: db,table
  # Sets the initial schema.
  scan.value.initial-schemas.ddls: |
    CREATE TABLE db1.t1 (id BIGINT, name VARCHAR(10));
    CREATE TABLE db1.t2 (id BIGINT);
  # Sets the schema inference strategy to continuous. This strategy detects the schema of each message and synchronizes schema changes.
  schema.inference.strategy: continuous
  # Ignores errors during data parsing.
  ingestion.ignore-errors: true
  # If 100 data parsing errors occur, the job fails.
  ingestion.error-tolerance.max-count: 100
  
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
pipeline:
  # Enables dirty data collection. Dirty data is written to a log file.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

La configuration précédente spécifie un type initial BIGINT pour le champ id et VARCHAR(10) pour le champ name dans la table db1.t1. Elle spécifie également un type initial BIGINT pour le champ id dans la table db1.t2.

(JSON) Définir des types de champs fixes

Pour les données JSON, l'analyse des types de nœuds JSON permet d'inférer les types de champs. Si les types inférés ne correspondent pas à vos attentes, utilisez la configuration json.infer-schema.fixed-types pour spécifier les types de certains champs.

La configuration de job suivante spécifie le type du champ id comme BIGINT et le type du champ name comme VARCHAR(10).

source:
  type: kafka
  name: Kafka source
  properties.bootstrap.servers: localhost:9092
  topic: test_topic
  properties.group.id: test_group
  scan.startup.mode: earliest-offset
  value.format: json
  # Sets a fixed type for specific fields.
  json.infer-schema.fixed-types: id BIGINT, name VARCHAR(10)
  # Required for versions 11.5 and earlier.
  scan.max.pre.fetch.records: 0
  # Sets the schema inference strategy to continuous. This strategy detects the schema of each message and synchronizes schema changes.
  schema.inference.strategy: continuous
  # Ignores errors during data parsing.
  ingestion.ignore-errors: true
  # If 100 data parsing errors occur, the job fails.
  ingestion.error-tolerance.max-count: 100

# Writes all data from the test_topic topic to the test_database.test_topic table.
route:
  - source-table: test_topic
    sink-table: test_database.test_topic
    
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
pipeline:
  # Enables dirty data collection. Dirty data is written to a log file.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

(Canal JSON) Spécifier la source d'inférence

Les données Canal JSON contiennent plus d'informations que les données JSON standard. Si les champs sqlType ou mysqlType existent dans les données Canal JSON, vous pouvez utiliser ces informations pour analyser des types de données plus précis.

Analyse du schéma à partir de sqlType

source:
  type: kafka
  name: Kafka source
  properties.bootstrap.servers: localhost:9092
  topic: test_topic
  properties.group.id: test_group
  scan.startup.mode: earliest-offset
  value.format: canal-json
  # Infers schema from the sqlType field.
  canal-json.infer-schema.strategy: SQL_TYPE
  # Sets the schema inference strategy to continuous. This strategy detects the schema of each message and synchronizes schema changes.
  schema.inference.strategy: continuous
  # Ignores errors during data parsing.
  ingestion.ignore-errors: true
  # If 100 data parsing errors occur, the job fails.
  ingestion.error-tolerance.max-count: 100
  
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
pipeline:
  # Enables dirty data collection. Dirty data is written to a log file.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Analyse du schéma à partir de mysqlType

source:
  type: kafka
  name: Kafka source
  properties.bootstrap.servers: localhost:9092
  topic: test_topic
  properties.group.id: test_group
  scan.startup.mode: earliest-offset
  value.format: canal-json
  # Infers schema from the mysqlType field.
  canal-json.infer-schema.strategy: MYSQL_TYPE
  # Sets the schema inference strategy to continuous. This strategy detects the schema of each message and synchronizes schema changes.
  schema.inference.strategy: continuous
  # Ignores errors during data parsing.
  ingestion.ignore-errors: true
  # If 100 data parsing errors occur, the job fails.
  ingestion.error-tolerance.max-count: 100
  
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
pipeline:
  # Enables dirty data collection. Dirty data is written to a log file.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Utiliser un schéma statique

Si le schéma des données dans votre topic est fixe, définissez schema.inference.strategy sur static. Le job d'ingestion de données effectue alors l'inférence de schéma une seule fois au démarrage et n'analyse pas le schéma des données ultérieures.

source:
  type: kafka
  properties.bootstrap.servers: localhost:9092
  topic: test_topic
  properties.group.id: test_group
  scan.startup.mode: earliest-offset
  value.format: json
  # Sets the schema inference strategy to static. The schema is inferred only once when the job starts.
  schema.inference.strategy: static
  # Tries to consume 20 records from each partition to infer the schema.
  scan.max.pre.fetch.records: 20
  # Ignores errors during data parsing.
  ingestion.ignore-errors: true
  # If 100 data parsing errors occur, the job fails.
  ingestion.error-tolerance.max-count: 100
  
# Writes all data from the test_topic topic to the test_database.test_topic table.
route:
  - source-table: test_topic
    sink-table: test_database.test_topic
    
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
pipeline:
  # Enables dirty data collection. Dirty data is written to a log file.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Vous pouvez également ajouter le paramètre scan.value.initial-schemas.ddls pour spécifier un schéma initial et ignorer l'inférence de schéma pour certaines tables. L'exemple suivant spécifie les schémas initiaux pour les tables db1.t1 et db1.t2.

source:
  type: kafka
  name: Kafka source
  properties.bootstrap.servers: localhost:9092
  topic: test_topic
  properties.group.id: test_group
  scan.startup.mode: earliest-offset
  value.format: json
  # Uses the db and table fields from the data as the TableID.
  json.decode.parser-table-id.fields: db,table
  # Sets the initial schema.
  scan.value.initial-schemas.ddls: | 
    CREATE TABLE db1.t1 (id BIGINT, name VARCHAR(10));
    CREATE TABLE db1.t2 (id BIGINT);
  # Sets the schema inference strategy to static. The schema is inferred only once when the job starts.
  schema.inference.strategy: static
  # Ignores errors during data parsing.
  ingestion.ignore-errors: true
  # If 100 data parsing errors occur, the job fails.
  ingestion.error-tolerance.max-count: 100
  
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
pipeline:
  # Enables dirty data collection. Dirty data is written to a log file.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Accélérer l'analyse des données de journal Kafka

Outre les configurations générales d'accélération du connecteur Kafka, le job d'ingestion de données Flink CDC propose des paramètres spécifiques pour accélérer l'analyse selon votre cas d'utilisation.

L'analyse du schéma peut prendre du temps. Si votre application en aval nécessite uniquement le type String, activez json.infer-schema.primitive-as-string, debezium-json.infer-schema.primitive-as-string ou canal-json.infer-schema.primitive-as-string. Cela ignore l'inférence de type et définit tous les types de champs sur String, ce qui accélère l'analyse.
Pour les données Canal JSON, utilisez canal-json.database.include et canal-json.table.include pour filtrer les données des tables inutiles.
Si le schéma des données ne change pas ou si vous n'avez pas besoin de l'évolution du schéma, remplacez schema.inference.strategy par static. Cela effectue l'inférence de schéma une seule fois au démarrage du job.

Ingestion de journaux SLS dans un lac de données

Log Service (SLS) est un service tout-en-un pour les données de journal. L'ingestion de données Flink CDC vous permet d'ingérer rapidement des données de journal depuis SLS dans un lac de données en temps réel. Elle infère automatiquement le schéma et prend en charge son évolution.

source:
  type: sls
  endpoint: localhost
  project: test_pj
  logstore: test_log
  accessId: access_id
  accessKey: access_key
  
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
pipeline:
  # Enables dirty data collection. Dirty data is written to a log file.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

TableID : Par défaut, le project et le logstore sont concaténés pour former le TableID. Dans le job ci-dessus, le TableID est test_pj.test_log.
Type de données : Le connecteur SLS traite par défaut tous les champs de chaque entrée de journal comme le type String.
Évolution du schéma : L'évolution du schéma est actuellement limitée à l'ajout de nouvelles colonnes. Les nouvelles colonnes sont ajoutées à la fin du schéma par défaut.

Cas d'utilisation

Les sections suivantes décrivent les configurations de job pour des cas d'utilisation courants. Pour des configurations plus détaillées, consultez Connecteur SLS d'ingestion de données.

Lire les métadonnées

Utilisez la configuration metadata.list pour lire et transmettre des métadonnées SLS supplémentaires. Les colonnes de métadonnées ajoutées dans metadata.list peuvent être utilisées directement dans le module transform. Pour obtenir la liste des métadonnées prises en charge, consultez metadata.list.

Notez que les colonnes ajoutées avec metadata.list ne sont pas automatiquement incluses dans les données de sortie. Pour écrire ces colonnes de métadonnées dans la destination, déclarez-les dans la section projection du module transform.

La configuration suivante ajoute les métadonnées __timestamp__ et __tag__ aux données et les écrit dans la destination. Elle utilise également ces champs de métadonnées dans le module transform pour filtrer les données dont l'horodatage est supérieur à 1772181154 et le tag est "test".

source:
  type: sls
  endpoint: localhost
  project: test_pj
  logstore: test_log
  accessId: access_id
  accessKey: access_key
  # Adds metadata columns.
  metadata.list: __timestamp__,__tag__
  # Sets the schema inference strategy to continuous. This strategy detects the schema of each message and synchronizes schema changes.
  schema.inference.strategy: continuous
  # Ignores errors during data parsing.
  ingestion.ignore-errors: true
  # If 100 data parsing errors occur, the job fails.
  ingestion.error-tolerance.max-count: 100
  
transform:
  - source-table: \.*.\.*
    projection: \*, __timestamp__ as timestamp_col, __tag__ as tag_col
    filter: '`__timestamp__` > 1772181154 and `__tag__` = "test"'

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
pipeline:
  # Enables dirty data collection. Dirty data is written to a log file.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Gérer les données incorrectes

Les données de journal peuvent contenir des données incorrectes mal formatées, ce qui peut entraîner l'échec et le redémarrage répété du job. L'ingestion de données Flink CDC permet d'ignorer les erreurs d'analyse et de collecter les données ayant provoqué les erreurs. Pour plus d'informations, consultez Collecte de données incorrectes.

Le job suivant écrit dans un fichier journal les données incorrectes dont l'analyse a échoué. Le job échoue si plus de 100 erreurs d'analyse se produisent.

source:
  type: sls
  endpoint: localhost
  project: test_pj
  logstore: test_log
  accessId: access_id
  accessKey: access_key
  # Sets the schema inference strategy to continuous. This strategy detects the schema of each message and synchronizes schema changes.
  schema.inference.strategy: continuous
  # Ignores errors during data parsing.
  ingestion.ignore-errors: true
  # If 100 data parsing errors occur, the job fails.
  ingestion.error-tolerance.max-count: 100

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enables deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
pipeline:
  # Enables dirty data collection. Dirty data is written to a log file.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Spécifier un TableID

Pour utiliser des champs spécifiques des données comme TableID, configurez decode.table-id.fields. Par exemple, étant donné les données de journal {"col0":"a", "col1":"b", "col2":"c"}, différentes configurations génèrent les TableIDs suivants :