Utilisation du connecteur Print - Realtime Compute for Apache Flink

Le connecteur Print écrit chaque ligne dans la sortie standard ou le flux d'erreur standard du TaskManager. Utilisez-le pendant le développement pour inspecter les résultats intermédiaires ou vérifier la sortie finale sans configurer un sink de production.

La sortie du connecteur Print apparaît dans les journaux du TaskManager, et non dans la console de l'interface web Flink. Définissez le niveau de journalisation du TaskManager sur INFO pour afficher la sortie.

Fonctionnalités prises en charge

Élément	Description
Type de table	Table de résultats et sink d'ingestion de données
Mode d'exécution	Mode batch et mode streaming
Format des données	N/A
Métrique	N/A
Type d'API	API SQL et API YAML d'ingestion de données
Mise à jour ou suppression des données dans une table sink	Pris en charge

Cas d'utilisation

Tests de développement : Ajoutez une table de résultats Print parallèlement à votre sink existant pour vérifier que les bonnes lignes circulent avant l'écriture dans un système de production.

SQL

Syntaxe

CREATE TABLE print_table (
  a INT,
  b VARCHAR
) WITH (
  'connector' = 'print',
  'logger' = 'true'
);

Pour reproduire le schéma d'une table existante sans redéfinir les colonnes, utilisez la clause LIKE :

CREATE TABLE print_table WITH ('connector' = 'print')
LIKE table_source (EXCLUDING ALL)

Paramètres

Paramètre	Description	Type de données	Obligatoire	Valeur par défaut
`connector`	Type de connecteur. Définissez la valeur sur `print`.	String	Oui	—
`logger`	Indique s'il faut afficher la sortie dans la console.	Boolean	Non	`false`
`print-identifier`	Libellé ajouté au début de chaque ligne de sortie. Utilisez cette option pour distinguer les résultats lorsque plusieurs sinks Print sont actifs.	String	Non	—
`sink.parallelism`	Parallélisme du sink Print.	Int	Non	Identique à celui du flux en amont

Ingestion de données

Pour les jobs YAML d'ingestion de données, utilisez le connecteur values avec print.enabled: true pour router la sortie vers les journaux ou les fichiers de sortie.

Syntaxe

source:
  type: xxx

sink:
  type: values
  name: Values Sink
  print.enabled: true

Paramètres

Paramètre	Description	Type de données	Obligatoire	Valeur par défaut
`type`	Type de connecteur sink. Définissez la valeur sur `values`.	STRING	Oui	—
`name`	Nom d'affichage du sink.	STRING	Non	—
`print.enabled`	Active le comportement d'impression. Définissez la valeur sur `true`.	BOOLEAN	Oui	—
`sink.print.standard-error`	Indique s'il faut écrire la sortie sur stderr au lieu de stdout.	BOOLEAN	Non	`false`
`sink.print.logger`	Indique s'il faut afficher la sortie dans la console.	BOOLEAN	Non	`false`
`sink.print.limit`	Nombre maximal d'enregistrements à imprimer.	LONG	Non	`2000`
`materialized.in.memory`	Indique s'il faut conserver les événements de journal binaire en mémoire.	BOOLEAN	Non	`false`
`error.on.schema.change`	Indique s'il faut générer une erreur en cas de changement de schéma.	BOOLEAN	Non	`false`

Exemples

Table de résultats SQL

L'exemple suivant lit les données d'une table source et écrit toutes les lignes dans un sink Print. Consultez les journaux du TaskManager pour voir la sortie.

CREATE TEMPORARY TABLE table_source (
  name VARCHAR,
  score BIGINT
) WITH (
  ...
);

CREATE TEMPORARY TABLE print_sink (
  name VARCHAR,
  score BIGINT
) WITH (
  'connector' = 'print'
);

INSERT INTO print_sink SELECT * FROM table_source;

Sink d'ingestion de données

L'exemple suivant ingère des données depuis MySQL et imprime toutes les lignes dans le fichier de sortie.

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 7601-7604

sink:
  type: values
  name: Values Sink
  print.enabled: true

Limitations

Niveau de journalisation : Définissez le niveau de journalisation du TaskManager sur INFO pour voir la sortie du connecteur Print. La sortie est écrite dans les journaux du TaskManager, et non dans la console de l'interface web Flink.
Limite d'enregistrements : TaskManager.out affiche un maximum de 2 000 entrées de journal. Pour inspecter des lignes spécifiques lorsque le volume de données est important, ajoutez une clause WHERE pour filtrer les lignes avant qu'elles n'atteignent le sink Print.