Écriture externe (Spark, Flink, Kafka)

Les moteurs externes peuvent écrire dans une table gérée Scrydon, et pas seulement la lire. Un job Spark Structured Streaming, un pipeline Flink ou un connecteur Iceberg sink de Kafka Connect effectue ses commits via la même façade de catalogue REST Iceberg que celle que vous utilisez pour les lectures — Scrydon gouverne le commit a posteriori plutôt que de le bloquer en ligne, de sorte que votre pipeline conserve le débit natif du moteur.

Cette page couvre l'écriture externe en flux et par lots. Pour les lectures ponctuelles et la configuration de la connexion propre à chaque moteur, consultez Connecter un moteur.

Quand utiliser l'écriture externe

Optez pour l'écriture externe lorsqu'un producteur à fort volume parle déjà Iceberg et que vous souhaitez qu'il écrive directement dans une table gouvernée :

• Spark Structured Streaming / batch — ajouts par micro-lots ou en masse.
• Apache Flink — ajouts continus en flux.
• Kafka Connect avec le connecteur Iceberg sink — topic → table.

Si vous souhaitez plutôt que chaque ligne soit classifiée et auditée de manière synchrone à son arrivée, utilisez le plan de données gouverné (tables.write du SDK de notebook scrydon, ou POST /data/v1/tables/:ref/write) — et non l'écriture externe.

Prérequis

Configurer l'écrivain

Pointez le moteur vers le catalogue de votre organisation exactement comme pour les lectures — le chemin d'écriture utilise la même URI de catalogue, le même entrepôt et le même jeton. Définissez le jeton d'écriture dans la configuration du catalogue REST Iceberg du moteur.

spark = (
    SparkSession.builder
    .config("spark.sql.catalog.scrydon", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.scrydon.type", "rest")
    .config("spark.sql.catalog.scrydon.uri", "https://<your-scrydon-host>/api/table/iceberg")
    .config("spark.sql.catalog.scrydon.token", "<WRITE_ENGINE_TOKEN>")
    .getOrCreate()
)

# Micro-batch append into an open_append table.
(
    streaming_df.writeStream
    .format("iceberg")
    .outputMode("append")
    .option("path", "scrydon.<namespace>.<table>")
    .option("checkpointLocation", "/chk/scrydon-sink")
    .trigger(processingTime="30 seconds")
    .start()
)

CREATE CATALOG scrydon WITH (
  'type' = 'iceberg',
  'catalog-type' = 'rest',
  'uri' = 'https://<your-scrydon-host>/api/table/iceberg',
  'token' = '<WRITE_ENGINE_TOKEN>'
);

-- Streaming INSERT appends continuously into the open_append table.
INSERT INTO scrydon.`<namespace>`.`<table>`
SELECT * FROM source_stream;

{
  "name": "scrydon-iceberg-sink",
  "config": {
    "connector.class": "org.apache.iceberg.connect.IcebergSinkConnector",
    "topics": "events",
    "iceberg.catalog.type": "rest",
    "iceberg.catalog.uri": "https://<your-scrydon-host>/api/table/iceberg",
    "iceberg.catalog.token": "<WRITE_ENGINE_TOKEN>",
    "iceberg.tables": "<namespace>.<table>",
    "iceberg.tables.auto-create-enabled": "false"
  }
}

L'URI du catalogue ne doit pas inclure de suffixe /v1 — les clients REST Iceberg ajoutent eux-mêmes /v1/.

Comment les commits sont réconciliés

L'écriture externe est gouvernée a posteriori. Chaque commit externe arrive sous la forme d'un snapshot Iceberg normal ; le réconciliateur de commits externes de Scrydon, lors de son prochain passage (ou sur un événement de commit lorsque c'est configuré) :

1. Détecte le nouveau snapshot (écart par rapport au dernier snapshot enregistré par Scrydon).
2. Recompte les lignes sous gouvernance et rafraîchit le row_count de la table.
3. Réapplique les propriétés de gouvernance (scrydon.classification, scrydon.write-policy, colonnes d'identité, sensibilité des colonnes) sur la table.
4. Émet un événement d'audit table.external_commit_reconciled décrivant ce que le commit a ajouté (opération, lignes, fichiers, octets).

Vos ajouts sont donc immédiatement visibles pour les lecteurs Iceberg, et deviennent pleinement attribués dans la gouvernance et la piste d'audit peu après.

Diffusion des changements de source

Lorsque le bus d’événements est activé, Scrydon enregistre aussi des notifications privées pour les changements validés sur les tables Iceberg gérées actives. Un balayage distinct des métadonnées récupère les notifications de catalogue manquées, y compris après un redémarrage de service. Il s’exécute cinq secondes après la fin du balayage précédent ; la durée du balayage et le retard de diffusion influencent donc le temps de récupération. Les écritures gérées par la plateforme utilisent le même mécanisme.

Ces notifications contiennent des versions de source, et non des lignes de table, et n’accordent aucun accès aux données. Ontology conserve la dernière version reçue d’un redémarrage à l’autre et ignore les notifications en double ou plus anciennes. Cela prépare les requêtes d’ontologie gouvernées aux mises à jour en direct ; leur seule présence ne certifie ni une application Horizon en direct ni la fraîcheur d’un flux amont.

Comportement en cas de conflit avec les upserts de la plateforme

Le protocole de commit à concurrence optimiste d'Iceberg arbitre les écrivains concurrents : chaque commit est validé par rapport aux métadonnées actuelles de la table et retenté en cas de conflit. Un ajout externe (nouveaux fichiers, aucune suppression) et un upsert merge-on-read de la plateforme (suppressions + nouveaux fichiers) touchent des métadonnées disjointes dans le cas courant, si bien que les deux sont validés ; un conflit réel oblige le perdant à retenter sur le snapshot du gagnant. Aucun commit n'est perdu.

Relire les résultats

Lisez les lignes écrites via le plan de données gouverné pour que le masquage par colonne et le contrôle d'habilitation soient appliqués aux données elles-mêmes :

• Le SDK de notebook scrydon — tables.read(...).
• POST /data/v1/tables/:ref/query — masqué, paginé par clé.
• L'aperçu de table d'Analytics.

Pour une lecture Iceberg épinglée, envoyez asOf dans le corps de la requête gouvernée : { "asOf": { "snapshotId": "12345678901234567890" } } ou { "asOf": { "timestamp": "2024-01-01T00:00:00Z" } }. Utilisez exactement un sélecteur. L'horodatage est évalué en UTC. Sur chaque page du curseur, renvoyez le même identifiant physique de table, sort, filter et asOf avec le curseur retourné ; le curseur ne porte pas la sélection du snapshot. Un snapshot absent fait échouer la lecture au lieu de se replier sur les lignes actuelles.

Un scan externe brut lit les fichiers Parquet tels qu'ils sont stockés (sans masques par colonne) et il est soumis au contrôle d'habilitation au moment de la distribution des identifiants — voir Comment la gouvernance tient en fédération.