Gérer les flux

L'API Datastream vous permet d'automatiser vos workflows de réplication de données via des appels d'API REST ou la Google Cloud CLI. Vous pouvez créer, valider, modifier et supprimer des flux, ou récupérer et lister des flux et des objets de flux.

Vous pouvez également démarrer, suspendre, reprendre ou récupérer des flux, gérer le backfill d'objets, ajuster la simultanéité des tâches et diffuser des objets Oracle volumineux. Pour en savoir plus sur la commande, consultez gcloud datastream streams. Pour obtenir un tutoriel complet, consultez Utiliser les API Datastream.

Créer un flux

Dans cette section, vous allez apprendre à créer un flux permettant de transférer des données de votre source vers une destination. Les exemples suivants ne sont pas exhaustifs, mais mettent en évidence des fonctionnalités spécifiques de Datastream. Pour répondre à votre cas d'utilisation spécifique, utilisez ces exemples avec la documentation de référence de l'API Datastream.

Cette section couvre les cas d'utilisation suivants :

Exemple 1 : Diffuser des objets spécifiques vers BigQuery

Dans cet exemple, vous allez apprendre à :

  • Diffuser des données en streaming depuis MySQL vers BigQuery
  • Inclure un ensemble d'objets dans le flux
  • Définir le mode d'écriture du flux sur "append-only" (ajout uniquement)
  • Effectuer un backfill de tous les objets inclus dans le flux

Voici une requête permettant d'extraire toutes les tables de schema1 et deux tables spécifiques de schema2 : tableA et tableC. Les événements sont écrits dans un ensemble de données dans BigQuery.

La requête n'inclut pas le paramètre customerManagedEncryptionKey. Par conséquent, le système de gestion de clés interne Google Cloud est utilisé pour chiffrer vos données au lieu de CMEK.

Le paramètre backfillAll associé à l'exécution du backfill (ou instantané) historique est défini sur un dictionnaire vide ({}), ce qui signifie que Datastream effectue le backfill des données historiques de toutes les tables incluses dans le flux.

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=mysqlCdcStream
{
  "displayName": "MySQL CDC to BigQuery",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "schema1" },
          {
            "database": "schema2",
            "mysqlTables": [
              {
                "table": "tableA",
                "table": "tableC"
              }
            ]
          }
        ]
      },
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
          "location": "us",
          "datasetIdPrefix": "prefix_"
        }
      },
      "dataFreshness": "900s"
    }
  },
  "backfillAll": {}
}

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour créer un flux, consultez la documentation du SDK Google Cloud.

Exemple 2 : Exclure des objets spécifiques d'un flux avec une source PostgreSQL

Dans cet exemple, vous allez apprendre à :

  • Diffuser des données en streaming depuis PostgreSQL vers BigQuery
  • Exclure des objets du flux
  • Exclure des objets du remplissage

Le code suivant montre une requête permettant de créer un flux utilisé pour transférer des données d'une base de données PostgreSQL source vers BigQuery. Lorsque vous créez un flux à partir d'une base de données PostgreSQL source, vous devez spécifier deux champs supplémentaires spécifiques à PostgreSQL dans votre requête :

  • replicationSlot : un emplacement de réplication est un prérequis pour configurer une base de données PostgreSQL pour la réplication. Vous devez créer un emplacement de réplication pour chaque flux.
  • publication : une publication est un groupe de tables dont vous souhaitez répliquer les modifications. Le nom de la publication doit exister dans la base de données avant de démarrer un flux. La publication doit au moins inclure les tables spécifiées dans la liste includeObjects du flux.

Le paramètre backfillAll associé à l'exécution du backfill (ou instantané) historique est défini pour exclure une table.

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams?streamId=myPostgresStream
{
  "displayName": "PostgreSQL to BigQueryCloud Storage",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/connectionProfiles/postgresCp",
    "postgresqlSourceConfig": {
      "replicationSlot": "replicationSlot1",
      "publication": "publicationA",
      "includeObjects": {
        "postgresqlSchemas": {
          "schema": "schema1"
        }
      },
      "excludeObjects": {
        "postgresqlSchemas": [
          { "schema": "schema1",
        "postgresqlTables": [
          {
            "table": "tableA",
            "postgresqlColumns": [
              { "column": "column5" }
              ]
              }
            ]
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "dataFreshness": "900s",
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
           "location": "us",
           "datasetIdPrefix": "prefix_"
        }
      }
    }
  },
  "backfillAll": {
    "postgresqlExcludedObjects": {
        "postgresqlSchemas": [
          { "schema": "schema1",
            "postgresqlTables": [
              { "table": "tableA" }
            ]
          }
        ]
      }
    }
  }

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour créer un flux, consultez la documentation du SDK Google Cloud.

Exemple 3 : Spécifier le mode d'écriture "append-only" (ajout uniquement) pour un flux

Lorsque vous diffusez des données vers BigQuery, vous pouvez définir le mode d'écriture : merge ou appendOnly. Pour en savoir plus, consultez Configurer le mode d'écriture.

Si vous ne spécifiez pas le mode d'écriture dans votre requête de création de flux, le mode merge par défaut est utilisé.

La requête suivante montre comment définir le mode appendOnly lorsque vous créez un flux MySQL vers BigQuery.

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=appendOnlyStream
{
  "displayName": "My append-only stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "myMySqlDb"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
          "location": "us",
          "datasetIdPrefix": "prefix_"
        }
      },
      "appendOnly": {}
    }
  },
  "backfillAll": {}
}

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour créer un flux, consultez la documentation du SDK Google Cloud.

Exemple 4 : Diffuser vers un autre projet dans BigQuery

Si vous avez créé vos ressources Datastream dans un projet, mais que vous souhaitez les diffuser dans un autre projet BigQuery, vous pouvez le faire à l'aide d'une requête semblable à celle-ci.

Si vous spécifiez sourceHierarchyDatasets pour votre ensemble de données de destination, vous devez renseigner le champ projectId.

Si vous spécifiez singleTargetDataset pour votre ensemble de données de destination, renseignez le champ datasetId au format projectId:datasetId.

REST

Pour sourceHierarchyDatasets :

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=crossProjectBqStream1
{
  "displayName": "My cross-project stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "myMySqlDb"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
          "location": "us",
          "datasetIdPrefix": "prefix_"
        },
        "projectId": "myProjectId2"
      }
    }
  },
  "backfillAll": {}
}

Pour singleTargetDataset :

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=crossProjectBqStream2
{
  "displayName": "My cross-project stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "myMySqlDb"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "singleTargetDataset": {
        "datasetId": "myProjectId2:myDatasetId"
      },
    }
  },
  "backfillAll": {}
}

gcloud

Pour sourceHierarchyDatasets :

  datastream streams create crossProjectBqStream1 --location=us-central1
  --display-name=my-cross-project-stream --source=source-cp --mysql-source-config=mysql_source_config.json
  --destination=destination-cp --bigquery-destination-config=source_hierarchy_cross_project_config.json
  --backfill-none

Contenu du fichier de configuration source_hierarchy_cross_project_config.json :

  {"sourceHierarchyDatasets": {"datasetTemplate": {"location": "us-central1", "datasetIdPrefix": "prefix_"}, "projectId": "myProjectId2"}}

Pour singleTargetDataset :

  datastream streams create crossProjectBqStream --location=us-central1
  --display-name=my-cross-project-stream --source=source-cp --mysql-source-config=mysql_source_config.json
  --destination=destination-cp --bigquery-destination-config=single_target_cross_project_config.json
  --backfill-none

Contenu du fichier de configuration single_target_cross_project_config.json :

  {"singleTargetDataset": {"datasetId": "myProjectId2:myDatastetId"}}

Pour en savoir plus sur l'utilisation de gcloud pour créer un flux, consultez la documentation du SDK Google Cloud.

Exemple 5 : Diffuser vers une destination Cloud Storage

Dans cet exemple, vous allez apprendre à :

  • Transférer des données d'Oracle vers Cloud Storage
  • Définir un ensemble d'objets à inclure dans le flux
  • Définir une CMEK pour chiffrer les données au repos

La requête suivante montre comment créer un flux qui écrit les événements dans un bucket Cloud Storage.

Dans cet exemple de requête, les événements sont écrits au format de sortie JSON, et un nouveau fichier est créé tous les 100 Mo ou toutes les 30 secondes (en remplaçant les valeurs par défaut de 50 Mo et 60 secondes).

Pour le format JSON, vous pouvez :

  • Incluez un fichier de schéma de types unifiés dans le chemin d'accès. Par conséquent, Datastream écrit deux fichiers dans Cloud Storage : un fichier de données JSON et un fichier de schéma Avro. Le fichier de schéma porte le même nom que le fichier de données, avec une extension .schema.

  • Activez la compression gzip pour que Datastream compresse les fichiers écrits dans Cloud Storage.

En utilisant le paramètre backfillNone, la requête spécifie que seules les modifications en cours sont diffusées dans la destination, sans backfill.

La requête spécifie le paramètre de clé de chiffrement gérée par le client, qui vous permet de contrôler les clés utilisées pour chiffrer les données au repos dans un projet Google Cloud . Le paramètre fait référence à la CMEK que Datastream utilise pour chiffrer les données diffusées de la source vers la destination. Il spécifie également le trousseau de clés pour votre CMEK.

Pour en savoir plus sur les trousseaux de clés, consultez Ressources Cloud KMS. Pour en savoir plus sur la protection de vos données à l'aide de clés de chiffrement, consultez Cloud Key Management Service (KMS).

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams?streamId=myOracleCdcStream
{
  "displayName": "Oracle CDC to Cloud Storage",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/
    connectionProfiles/OracleCp",
    "oracleSourceConfig": {
      "includeObjects": {
        "oracleSchemas": [
          {
            "schema": "schema1"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "GcsBucketCp",
    "gcsDestinationConfig": {
      "path": "/folder1",
      "jsonFileFormat": {
        "schemaFileFormat": "AVRO_SCHEMA_FILE"
      },
      "fileRotationMb": 100,
      "fileRotationInterval": 30
    }
  },
  "customerManagedEncryptionKey": "projects/myProjectId1/locations/us-central1/
  keyRings/myRing/cryptoKeys/myEncryptionKey",
  "backfillNone": {}
}

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour créer un flux, consultez la documentation du SDK Google Cloud.

Exemple 6 : Diffuser des données dans une table Apache Iceberg

Dans cet exemple, vous allez apprendre à configurer un flux pour répliquer des données d'une base de données MySQL vers une table Apache Iceberg en mode append-only. Avant de créer la requête, assurez-vous d'avoir effectué les étapes suivantes :

  • Disposer d'un bucket Cloud Storage dans lequel vous souhaitez stocker vos données
  • Créer une connexion aux ressources Cloud
  • Accorder à votre connexion aux ressources Cloud l'accès au bucket Cloud Storage

Vous pouvez ensuite utiliser la requête suivante pour créer votre flux :

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=mysqlIcebergStream
{
  "displayName": "MySQL to Apache Iceberg stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlIcebergCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          {
            "database": "my-mysql-database"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "projects/myProjectId1/locations/us-central1/connectionProfiles/my-bq-cp-id",
    "bigqueryDestinationConfig": {
      "blmtConfig": {
        "bucket": "my-gcs-bucket-name",
        "rootPath": "my/folder",
        "connectionName": "my-project-id.us-central1.my-bigquery-connection-name",
        "fileFormat": "PARQUET",
        "tableFormat": "ICEBERG"
        },
      "singleTargetDataset": {
        "datasetId": "my-project-id:my-bigquery-dataset-id"
      },
      "appendOnly": {}
    }
  },
  "backfillAll": {}
}

gcloud

datastream streams create mysqlIcebergStream --location=us-central1
--display-name=mysql-to-bl-stream --source=source --mysql-source-config=mysql_source_config.json
--destination=destination --bigquery-destination-config=bl_config.json
--backfill-none

Contenu du fichier de configuration source mysql_source_config.json :

{"excludeObjects": {}, "includeObjects": {"mysqlDatabases":[{"database":"my-mysql-database"}]}}

Contenu du fichier de configuration bl_config.json :

{ "blmtConfig": { "bucket": "my-gcs-bucket-name", "rootPath": "my/folder", "connectionName": "my-project-id.us-central1.my-bigquery-connection-name", "fileFormat": "PARQUET", "tableFormat": "ICEBERG" }, "singleTargetDataset": {"datasetId": "my-project-id:my-bigquery-dataset-id"}, "appendOnly": {} }

Terraform

resource "google_datastream_stream" "stream" {
  stream_id    = "mysqlBlStream"
  location     = "us-central1"
  display_name = "MySQL to Apache Iceberg stream"

  source_config {
    source_connection_profile = "/projects/myProjectId1/locations/us-central1/streams/mysqlBlCp"
    mysql_source_config {
      include_objects {
        mysql_databases {
          database = "my-mysql-database"
        }
      }
    }
  }

  destination_config {
    destination_connection_profile = "projects/myProjectId1/locations/us-central1/connectionProfiles/my-bq-cp-id"
    bigquery_destination_config {
      single_target_dataset {
        dataset_id = "my-project-id:my-bigquery-dataset-id"
      }
      blmt_config {
        bucket          = "my-gcs-bucket-name"
        table_format    = "ICEBERG"
        file_format     = "PARQUET"
        connection_name = "my-project-id.us-central1.my-bigquery-connection-name"
        root_path       = "my/folder"
      }
      append_only {}
    }
  }

  backfill_none {}
}

Exemple 7 : Diffuser des données en streaming depuis MongoDB en mode JSON strict

Dans cet exemple, vous allez configurer un flux pour répliquer des données d'une source MongoDB vers BigQuery en mode JSON strict.

Par défaut, Datastream utilise le mode JSON canonique (CANONICAL) pour donner la priorité à la fidélité des données. Si vos applications en aval nécessitent une compatibilité avec les analyseurs JSON standards, vous pouvez spécifier le mode JSON strict (STRICT).

Lorsque vous créez un flux à partir d'une source MongoDB :

  • Les valeurs valides pour jsonMode dans l'API sont CANONICAL (par défaut) et STRICT.
  • Dans la gcloud CLI, il n'existe pas d'option autonome pour le mode JSON. Configurez plutôt les paramètres de la source MongoDB via un fichier de configuration transmis à l'option --mongodb-source-config.

REST

Dans l'API REST Datastream, définissez jsonMode sur "STRICT" dans l'objet sourceConfig.mongodbSourceConfig :

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams?streamId=STREAM_ID
{
  "displayName": "STREAM_DISPLAY_NAME",
  "sourceConfig": {
    "sourceConnectionProfile": "projects/PROJECT_ID/locations/LOCATION/connectionProfiles/SOURCE_PROFILE",
    "mongodbSourceConfig": {
      "jsonMode": "STRICT",
      "includeObjects": {
        "databases": [
          {
            "database": "DATABASE_NAME",
            "collections": [
              {
                "collection": "COLLECTION_NAME"
              }
            ]
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfile": "projects/PROJECT_ID/locations/LOCATION/connectionProfiles/DEST_PROFILE",
    "bigqueryDestinationConfig": {
      "singleTargetDataset": {
        "datasetId": "PROJECT_ID:DATASET_ID"
      }
    }
  }
}

Remplacez les éléments suivants :

  • PROJECT_ID : ID de votre projet Google Cloud .
  • LOCATION : région dans laquelle le flux est créé, par exemple us-central1.
  • STREAM_ID : identifiant du flux.
  • STREAM_DISPLAY_NAME : nom convivial du flux.
  • SOURCE_PROFILE : identifiant du profil de connexion MongoDB source.
  • DEST_PROFILE : identifiant du profil de connexion de destination.
  • DATABASE_NAME : nom de la base de données dans le cluster MongoDB source.
  • COLLECTION_NAME : nom de la collection dans la base de données.
  • DATASET_ID : identifiant de l'ensemble de données de destination dans BigQuery.

gcloud

Pour créer un flux avec une source MongoDB et le mode JSON strict à l'aide de la gcloud CLI :

  1. Créez un fichier de configuration source au format JSON ou YAML en spécifiant "jsonMode": "STRICT" et les objets à répliquer.

Fichier de configuration JSON, par exemple mongo_source_config.json :

{
  "jsonMode": "STRICT",
  "includeObjects": {
    "databases": [
      {
        "database": "DATABASE_NAME",
        "collections": [
          {
            "collection": "COLLECTION_NAME"
          }
        ]
      }
    ]
  }
}

Fichier de configuration YAML, par exemple mongo_source_config.yaml :

jsonMode: STRICT
includeObjects:
  databases:
    - database: DATABASE_NAME
      collections:
        - collection: COLLECTION_NAME
  1. Transmettez le fichier de configuration à l'option --mongodb-source-config lors de la création du flux :
gcloud datastream streams create STREAM_ID \
    --location=LOCATION \
    --display-name="STREAM_DISPLAY_NAME" \
    --source=projects/PROJECT_ID/locations/LOCATION/connectionProfiles/SOURCE_PROFILE \
    --destination=projects/PROJECT_ID/locations/LOCATION/connectionProfiles/DEST_PROFILE \
    --mongodb-source-config=mongo_source_config.json \
    --bigquery-destination-config=bigquery_destination_config.json

Contenu du fichier de configuration bigquery_destination_config.json :

{
  "singleTargetDataset": {
    "datasetId": "PROJECT_ID:DATASET_ID"
  }
}

Remplacez les éléments suivants :

  • STREAM_ID : identifiant du flux.
  • LOCATION : région dans laquelle le flux est créé, par exemple us-central1.
  • STREAM_DISPLAY_NAME : nom convivial du flux.
  • PROJECT_ID : ID de votre projet Google Cloud .
  • SOURCE_PROFILE : identifiant du profil de connexion MongoDB source.
  • DEST_PROFILE : identifiant du profil de connexion de destination.
  • DATABASE_NAME : nom de la base de données dans le cluster MongoDB source.
  • COLLECTION_NAME : nom de la collection dans la base de données.
  • DATASET_ID : identifiant de l'ensemble de données de destination dans BigQuery.

Pour en savoir plus sur l'utilisation de gcloud pour créer un flux, consultez la documentation du SDK Google Cloud.

Valider la définition d'un flux

Avant de créer un flux, vous pouvez valider sa définition. Vous pouvez ainsi vous assurer que toutes les vérifications de validation sont réussies et que le flux s'exécutera correctement une fois créé.

Valider un flux consiste à vérifier :

  • Si la source est correctement configurée pour autoriser Datastream à diffuser des données à partir de celle-ci.
  • Si le flux peut se connecter à la source et à la destination.
  • La configuration de bout en bout du flux.

Pour valider un flux, ajoutez &validate_only=true à l'URL précédant le corps de votre requête :

POST "https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams?streamId=STREAM_ID&validate_only=true"

Une fois cette requête envoyée, vous voyez les vérifications de validation exécutées par Datastream pour votre source et votre destination, ainsi que les réussites ou les échecs des contrôles. Si un contrôle échoue, des informations indiquent la raison de l'échec et la procédure à suivre pour corriger le problème.

Par exemple, supposons que vous disposiez d'une clé de chiffrement gérée par le client (CMEK) que vous souhaitez que Datastream utilise pour chiffrer les données diffusées de la source vers la destination. Lors de la validation du flux, Datastream vérifie que la clé existe et qu'il est autorisé à l'utiliser. Si l'une de ces conditions n'est pas remplie, le message d'erreur suivant s'affiche lorsque vous validez le flux :

CMEK_DOES_NOT_EXIST_OR_MISSING_PERMISSIONS

Pour résoudre ce problème, vérifiez que la clé que vous avez fournie existe et que le compte de service Datastream dispose de l'autorisation cloudkms.cryptoKeys.get pour la clé.

Une fois les corrections appropriées effectuées, relancez la requête pour vous assurer que tous les contrôles de validation réussissent. Dans l'exemple précédent, la vérification CMEK_VALIDATE_PERMISSIONS ne renverra plus de message d'erreur, mais aura l'état PASSED.

Obtenir des informations sur un flux

Le code suivant montre une requête permettant de récupérer des informations sur un flux. Voici quelques exemples :

  • Nom du flux (identifiant unique)
  • Nom convivial pour le flux (nom à afficher)
  • Codes temporels de création et de dernière mise à jour du flux
  • Informations sur les profils de connexion source et de destination associés au flux
  • État du flux

REST

GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams/STREAM_ID

La réponse s'affiche comme suit :

{
  "name": "myOracleCdcStream",
  "displayName": "Oracle CDC to Cloud Storage",
  "createTime": "2019-12-15T15:01:23.045123456Z",
  "updateTime": "2019-12-15T15:01:23.045123456Z",
  "sourceConfig": {
    "sourceConnectionProfileName": "myOracleDb",
    "oracleSourceConfig": {
      "includeObjects": {
        "oracleSchemas": [
          {
            "schema": "schema1"
          },
          {
            "schema": "schema3",
            "oracleTables": [
              { "table": "tableA" },
              { "table": "tableC" }
            ]
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "myGcsBucket",
    "gcsDestinationConfig": {
      "path": "/folder1",
      "avroFileFormat": {},
      "fileRotationMb": 100,
      "fileRotationInterval": 60
    }
  },
  "state": "RUNNING"
  "customerManagedEncryptionKey": "projects/myProjectId1/locations/us-central1/
  keyRings/myRing/cryptoKeys/myEncryptionKey",
  "backfillAll": {}
}

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour récupérer des informations sur votre flux, consultez la documentation du SDK Google Cloud.

Répertorier les flux

Le code suivant montre une requête permettant de récupérer la liste de tous les flux du projet et de l'emplacement spécifiés.

REST

GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour récupérer des informations sur tous vos flux, consultez la documentation du SDK Google Cloud.

Lister les objets d'un flux

Le code suivant montre une requête permettant de récupérer des informations sur tous les objets d'un flux.

REST

GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID/objects

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour récupérer des informations sur tous les objets de votre flux, consultez la documentation du SDK Google Cloud.

La liste des objets renvoyés peut ressembler à ceci :

REST

{
  "streamObjects": [
    {
      "name": "projects/myProjectId1/locations/us-central1/streams/myStream/
      objects/object1",
      "displayName": "employees.salaries",
      "backfillJob": {
        "state": "ACTIVE",
        "trigger": "AUTOMATIC",
        "lastStartTime": "2021-10-18T12:12:26.344878Z"
      },
      "sourceObject": {
        "mysqlIdentifier": {
          "database": "employees",
          "table": "salaries"
        }
      }
    },
    {
      "name": "projects/myProjectId1/locations/us-central1/streams/myStream/
      objects/object2",
      "displayName": "contractors.hours",
      "sourceObject": {
        "mysqlIdentifier": {
          "database": "contractors",
          "table": "hours"
        }
      }
    },
    {
      "name": "projects/myProjectId1/locations/us-central1/streams/myStream/
      objects/object3",
      "displayName": "employees.departments",
      "backfillJob": {
        "state": "COMPLETED",
        "trigger": "AUTOMATIC",
        "lastStartTime": "2021-10-18T11:26:12.869880Z",
        "lastEndTime": "2021-10-18T11:26:28.405653Z"
      },
      "sourceObject": {
        "mysqlIdentifier": {
          "database": "employees",
          "table": "departments"
        }
      }
    }
  ]
}

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour lister les objets d'un flux, consultez la documentation du SDK Google Cloud.

Lancer un flux

Le code suivant montre une requête de démarrage d'un flux.

Si vous utilisez le paramètre updateMask dans la requête, seuls les champs que vous spécifiez doivent être inclus dans le corps de la requête. Pour démarrer un flux, remplacez la valeur du champ state par RUNNING au lieu de CREATED.

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=state
{
  "state": "RUNNING"
}

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour démarrer votre flux, consultez la documentation du SDK Google Cloud.

Suspendre un flux

Le code suivant montre une requête permettant de mettre en pause un flux en cours d'exécution.

Dans cet exemple, le champ spécifié pour le paramètre updateMask est le champ state. En mettant en pause le flux, vous faites passer son état de RUNNING à PAUSED.

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=state
{
  "state": "PAUSED"
}

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour suspendre votre flux, consultez la documentation du SDK Google Cloud.

Réactiver un flux

Le code suivant montre une requête permettant de reprendre un flux mis en pause.

Dans cet exemple, le champ spécifié pour le paramètre updateMask est le champ state. En réactivant le flux, vous faites passer son état de PAUSED à RUNNING.

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=state
{
  "state": "RUNNING"
}

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour reprendre votre flux, consultez la documentation du SDK Google Cloud.

Récupérer un flux

Vous pouvez récupérer un flux en échec permanent à l'aide de la méthode RunStream. Chaque type de base de données source possède sa propre définition des opérations de récupération de flux possibles. Pour en savoir plus, consultez Récupérer un flux.

Récupérer un flux pour une source MySQL ou Oracle

Les exemples de code suivants montrent des requêtes permettant de récupérer un flux pour une source MySQL ou Oracle à partir de différentes positions de fichier journal :

REST

Récupérer un flux à partir de la position actuelle. Il s'agit de l'option par défaut :

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run

Récupérez un flux à partir de la prochaine position disponible :

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "nextAvailableStartPosition": {}
  }
}

Récupérer un flux à partir de la position la plus récente :

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "mostRecentStartPosition": {}
  }
}

Récupérer un flux à partir d'une position spécifique (réplication basée sur le binlog MySQL) :

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "NAME_OF_THE_LOG_FILE"
        "logPosition": POSITION
      }
    }
  }
}

Remplacez les éléments suivants :

  • NAME_OF_THE_LOG_FILE : nom du fichier journal à partir duquel vous souhaitez récupérer votre flux
  • POSITION : position dans le fichier journal à partir de laquelle vous souhaitez récupérer votre flux. Si vous ne fournissez pas la valeur, Datastream récupère le flux à partir de l'en-tête du fichier.

Exemple :

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "binlog.001"
        "logPosition": 4
      }
    }
  }
}

Récupérez un flux à partir d'une position spécifique (réplication MySQL basée sur GTID) :

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlGtidPosition": {
        "gtidSet": "GTID_SET"
      }
    }
  }
}

Remplacez GTID_SET par un ou plusieurs GTID uniques ou plages de GTID à partir desquels vous souhaitez récupérer votre flux.

Exemple :

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlGtidPosition": {
        "gtidSet": "22cc56f5-3862-379a-9af5-051c59baef9d:1-561143685:561143688-591036613,b8d7df02-832b-32b9-bec7-2018806b76f6:1-3"
      }
    }
  }
}

Récupérer un flux à partir d'une position spécifique (Oracle) :

POST https://datastream.googleapis.com/v1/projects/[project-id]/locations/
[location]/streams/[stream-id]:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": scn
      }
    }
  }
}
 Remplacez scn par le numéro de modification du système (SCN) dans le fichier journal de rétablissement à partir duquel vous souhaitez récupérer votre flux. Ce champ est obligatoire.

Exemple :

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": 234234
      }
    }
  }
}

Pour en savoir plus sur les options de récupération disponibles, consultez Récupérer un flux.

gcloud

La récupération d'un flux à l'aide de gcloud n'est pas possible.

Récupérer un flux pour une source PostgreSQL

L'exemple de code suivant montre une requête de récupération d'un flux pour une source PostgreSQL. Pendant la récupération, le flux commence à lire à partir du premier numéro de séquence de journal (LSN) dans l'emplacement de réplication configuré pour le flux.

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "nextAvailableStartPosition": {}
  }
}

Exemple :

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run

Si vous souhaitez modifier l'emplacement de réplication, mettez d'abord à jour le flux avec le nouveau nom de l'emplacement de réplication :

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=sourceConfig.postgresqlSourceConfig.replicationSlot
{
  "sourceConfig": {
    "postgresqlSourceConfig": {
      "replicationSlot": "NEW_REPLICATION_SLOT_NAME"
    }
  }
}

gcloud

La récupération d'un flux à l'aide de gcloud n'est pas possible.

Récupérer un flux pour une source SQL Server

Les exemples de code suivants montrent des exemples de requêtes permettant de récupérer un flux pour une source SQL Server.

REST

Récupérer un flux à partir de la première position disponible :

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run

Exemple :

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/mySqlServerStreamId:run

Récupérez un flux à partir d'un numéro de séquence de journal de votre choix :

POST https://datastream.googleapis.com/v1/projects/[project-id]/locations/
[location]/streams/[stream-id]:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "sqlServerLsnPosition": {
        "lsn": lsn
      }
    }
  }
}

Exemple :

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/mySqlServerStreamId:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "sqlServerLsnPosition": {
        "lsn": 0000123C:0000BA78:0004
      }
    }
  }
}

gcloud

La récupération d'un flux à l'aide de gcloud n'est pas possible.

Démarrer ou reprendre un flux à partir d'une position spécifique

Vous pouvez démarrer un flux ou reprendre un flux mis en pause à partir d'une position spécifique pour les sources MySQL, Oracle et SQL Server. Cela peut être utile lorsque vous souhaitez effectuer un remplissage à l'aide d'un outil externe ou démarrer la CDC à partir d'une position que vous indiquez :

  • Pour une source MySQL, vous devez indiquer une position de binlog ou un ensemble de GTID.
  • Pour une source Oracle, vous devez indiquer un numéro de modification du système (SCN) dans le fichier journal de rétablissement.
  • Pour une source SQL Server, vous devez indiquer un numéro de séquence de journal (LSN) dans une table de modifications.

Le code suivant montre des exemples de requêtes permettant de démarrer ou de reprendre un flux déjà créé à partir d'une position spécifique.

Démarrez ou reprenez un flux à partir d'une position de binlog spécifique (MySQL) :

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "NAME_OF_THE_LOG_FILE"
        "logPosition": POSITION
      }
    }
  }
}

Remplacez les éléments suivants :

  • NAME_OF_THE_LOG_FILE : nom du fichier journal à partir duquel vous souhaitez démarrer votre flux.
  • POSITION : position dans le fichier journal à partir de laquelle vous souhaitez démarrer votre flux. Si vous ne fournissez pas de valeur, Datastream commence la lecture à partir du début du fichier.

Exemple :

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "binlog.001"
        "logPosition": 2
      }
    }
  }
}

gcloud

Il n'est pas possible de démarrer ou de reprendre un flux à partir d'une position spécifique à l'aide de gcloud. Pour en savoir plus sur l'utilisation de gcloud pour démarrer ou reprendre un flux, consultez la documentation du SDK Cloud.

Démarrez ou reprenez un flux à partir d'un ensemble GTID spécifique (MySQL) :

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlGtidPosition": {
        "gtidSet": "GTID_SET"
      }
    }
  }
}

Remplacez GTID_SET par un ou plusieurs GTID uniques ou plages de GTID à partir desquels vous souhaitez démarrer ou reprendre votre flux.

Exemple :

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlGtidPosition": {
        "gtidSet": "22cc56f5-3862-379a-9af5-051c59baef9d:1-561143685:561143688-591036613,b8d7df02-832b-32b9-bec7-2018806b76f6:3-7"
      }
    }
  }
}

gcloud

Il n'est pas possible de démarrer ou de reprendre un flux à partir d'une position spécifique à l'aide de gcloud. Pour en savoir plus sur l'utilisation de gcloud pour démarrer ou reprendre un flux, consultez la documentation du SDK Cloud.

Démarrer ou reprendre un flux à partir d'un numéro de modification du système spécifique dans le fichier journal de rétablissement (Oracle) :

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": scn
      }
    }
  }
}
 Remplacez scn par le numéro de modification du système (SCN) dans le fichier journal de rétablissement à partir duquel vous souhaitez démarrer votre flux. Ce champ est obligatoire.

Exemple :

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": 123123
      }
    }
  }
}

gcloud

Il n'est pas possible de démarrer ou de reprendre un flux à partir d'une position spécifique à l'aide de gcloud. Pour en savoir plus sur l'utilisation de gcloud pour démarrer un flux, consultez la documentation du SDK Cloud.

Démarrer ou reprendre un flux à partir d'un numéro de séquence de journal spécifique (SQL Server) :

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "sqlServerLsnPosition": {
        "lsn": "lsn"
      }
    }
  }
}

Remplacez lsn par le numéro de séquence du journal dans la table des modifications à partir duquel vous souhaitez démarrer votre flux. Ce champ est obligatoire.

Exemple :

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "sqlServerLsnPosition": {
        "lsn": "0000123C0000BA780004"
      }
    }
  }
}

Pour en savoir plus, consultez la documentation de référence de l'API Datastream.

gcloud

Il n'est pas possible de démarrer ou de reprendre un flux à partir d'une position spécifique à l'aide de gcloud. Pour en savoir plus sur l'utilisation de gcloud pour démarrer un flux, consultez la documentation du SDK Cloud.

Modifier un flux

Le code suivant montre une requête de mise à jour de la configuration de rotation d'un flux pour alterner le fichier toutes les 75 Mo ou 45 secondes.

Dans cet exemple, les champs spécifiés pour le paramètre updateMask incluent les champs fileRotationMb et fileRotationInterval, représentés respectivement par les options destinationConfig.gcsDestinationConfig.fileRotationMb et destinationConfig.gcsDestinationConfig.fileRotationInterval.

REST

PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=destinationConfig.gcsDestinationConfig.
fileRotationMb,destinationConfig.gcsDestinationConfig.fileRotationInterval
{
  "destinationConfig": {
    "gcsDestinationConfig": {
      "fileRotationMb": 75,
      "fileRotationInterval": 45
    }
  }
}

Le code suivant montre une requête visant à inclure un fichier de schéma Unified Types dans le chemin d'accès des fichiers que Datastream écrit dans Cloud Storage. Par conséquent, Datastream écrit deux fichiers : un fichier de données JSON et un fichier de schéma Avro.

Dans cet exemple, le champ spécifié est le champ jsonFileFormat, représenté par l'option destinationConfig.gcsDestinationConfig.jsonFileFormat.

PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=destinationConfig.gcsDestinationConfig.
jsonFileFormat
{
  "destinationConfig": {
    "gcsDestinationConfig": {
      "jsonFileFormat" {
        "schemaFileFormat": "AVRO_SCHEMA_FILE"
      }  
    }
  }
}

Le code suivant montre une requête pour que Datastream réplique des données existantes, en plus des modifications en cours apportées aux données, de la base de données source vers la destination.

La section oracleExcludedObjects du code indique les tables et les schémas dont le backfill est limité dans la destination.

Pour cet exemple, toutes les tables et tous les schémas seront remplis, à l'exception de la table tableA dans le schéma schema3.

PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=backfillAll
{
  "backfillAll": {
    "oracleExcludedObjects": {
      "oracleSchemas": [
        {
          "schema": "schema3",
          "oracleTables": [
            { "table": "tableA" }
          ]
        }
      ]
    }
  }
}

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour modifier votre flux, consultez la documentation du SDK Google Cloud.

Déclencher le remplissage d'un objet d'un flux

Un flux dans Datastream peut effectuer un remplissage avec des données d'historique tout en diffusant les modifications en cours par flux vers une destination. Les modifications en cours seront toujours diffusées d'une source vers une destination. Toutefois, vous pouvez spécifier si vous souhaitez que les données historiques soient diffusées en flux continu.

Si vous souhaitez que les données historiques soient diffusées de la source vers la destination, utilisez le paramètre backfillAll.

Datastream vous permet également de ne diffuser que les données historiques de tables de base de données spécifiques. Pour ce faire, utilisez le paramètre backfillAll et excluez les tables pour lesquelles vous ne souhaitez pas de données historiques.

Si vous souhaitez que seules les modifications en cours soient diffusées en continu vers la destination, utilisez le paramètre backfillNone. Si vous souhaitez ensuite que Datastream diffuse un instantané de toutes les données existantes de la source vers la destination, vous devez lancer manuellement le remplissage des objets contenant ces données.

Il peut également être utile de lancer le remplissage d'un objet lorsque les données ne sont pas synchronisées entre la source et la destination. Par exemple, un utilisateur peut supprimer par inadvertance des données dans la destination, auquel cas les données sont perdues. Dans ce cas, le remplissage de l'objet sert de "mécanisme de réinitialisation", car toutes les données sont insérées par flux dans la destination en une seule opération. Par conséquent, les données sont synchronisées entre la source et la destination.

Avant de pouvoir lancer le backfill pour un objet d'un flux, vous devez récupérer des informations sur l'objet.

Chaque objet possède un OBJECT_ID, qui l'identifie de manière unique. Vous utilisez OBJECT_ID pour lancer le backfill du flux.

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID/objects/OBJECT_ID:startBackfillJob

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour lancer le backfill d'un objet de votre flux, consultez la documentation du SDK Google Cloud.

Arrêter le remplissage pour un objet d'un flux

Après avoir lancé le remplissage d'un objet d'un flux, vous pouvez l'arrêter. Par exemple, si un utilisateur modifie un schéma de base de données, le schéma ou les données peuvent être corrompus. Vous ne voulez pas que ce schéma ou ces données soient diffusés vers la destination. Vous arrêtez donc le remplissage de l'objet.

Vous pouvez également arrêter le backfill d'un objet à des fins d'équilibrage de charge. Datastream peut exécuter plusieurs remplissages en parallèle. Cela peut ajouter une charge supplémentaire à la source. Si la charge est importante, arrêtez le remplissage pour chaque objet, puis lancez-le pour les objets, un par un.

Avant de pouvoir arrêter le backfill d'un objet d'un flux, vous devez envoyer une requête pour récupérer des informations sur tous les objets d'un flux. Chaque objet renvoyé possède un OBJECT_ID, qui l'identifie de manière unique. Vous utilisez OBJECT_ID pour arrêter le backfill du flux.

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID/objects/OBJECT_ID:stopBackfillJob

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour arrêter le backfill d'un objet de votre flux, consultez la documentation du SDK Google Cloud.

Modifier le nombre maximal de tâches CDC simultanées

Le code suivant montre comment définir sur 7 le nombre maximal de tâches de capture de données modifiées (CDC) simultanées pour un flux MySQL.

Dans cet exemple, le champ spécifié pour le paramètre updateMask est le champ maxConcurrentCdcTasks. En définissant sa valeur sur 7, vous remplacez le nombre maximal de tâches CDC simultanées par 7. Vous pouvez utiliser des valeurs comprises entre 0 et 50 (inclus). Si vous ne définissez pas la valeur ou si vous la définissez sur 0, le paramètre système par défaut de 5 tâches est défini pour le flux.

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=sourceConfig.mysqlSourceConfig.maxConcurrentCdcTasks
{
  "sourceConfig": {
    "mysqlSourceConfig": {
      "maxConcurrentCdcTasks": "7"
      }
    }  
}

gcloud

Pour en savoir plus sur l'utilisation de gcloud, consultez la documentation du SDK Google Cloud.

Modifier le nombre maximal de tâches de remplissage simultanées

Le code suivant montre comment définir sur 25 le nombre maximal de tâches de backfill simultanées pour un flux MySQL.

Dans cet exemple, le champ spécifié pour le paramètre updateMask est le champ maxConcurrentBackfillTasks. En définissant sa valeur sur 25, vous remplacez la valeur précédente du nombre maximal de tâches de backfill simultanées par 25. Vous pouvez utiliser des valeurs comprises entre 0 et 50 (inclus). Si vous ne définissez pas la valeur ou si vous la définissez sur 0, le paramètre système par défaut (16 tâches) est défini pour le flux.

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/
streams/STREAM_ID?updateMask=sourceConfig.mysqlSourceConfig.maxConcurrentBackfillTasks
{
  "sourceConfig": {
    "mysqlSourceConfig": {
      "maxConcurrentBackfillTasks": "25"
      }
    }  
}

gcloud

Pour en savoir plus sur l'utilisation de gcloud, consultez la documentation du SDK Google Cloud.

Activer le streaming d'objets volumineux pour les sources Oracle

Vous pouvez activer le streaming d'objets volumineux, tels que les objets binaires volumineux (BLOB), les objets volumineux de caractères (CLOB) et les objets volumineux de caractères nationaux (NCLOB) pour les flux avec des sources Oracle. L'option streamLargeObjects vous permet d'inclure des objets volumineux dans les flux nouveaux et existants. L'option est définie au niveau du flux. Vous n'avez pas besoin de spécifier les colonnes des types de données d'objets volumineux.

L'exemple suivant montre comment créer un flux qui vous permet de diffuser des objets volumineux.

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams?streamId=myOracleLobStream
{
  "displayName": "Oracle LOB stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/connectionProfiles/OracleCp",
    "oracleSourceConfig": {
      "includeObjects": {
        "oracleSchemas": [
          {
            "schema": "schema1",
            "oracleTables": [
              {
                "table": "tableA",
                "oracleColumns": [
                  {
                    "column": "column1,column2"
                  }
                ]
              }
            ]
          }
        ]
      },
      "excludeObjects": {},
      "streamLargeObjects": {}
    }
  }
}

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour mettre à jour un flux, consultez la documentation du SDK Google Cloud.

Supprimer un flux

Le code suivant montre une requête de suppression d'un flux.

REST

DELETE https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID

gcloud

Pour en savoir plus sur l'utilisation de gcloud pour supprimer votre flux, consultez la documentation du SDK Google Cloud.

Étapes suivantes