Daten nach Pub/Sub exportieren (Reverse ETL)

Mit Reverse Extract-Transform-Load (RETL) von BigQuery zu Pub/Sub können Sie Daten ereignisgesteuert für nachgelagerte Anwendungen und Dienste bereitstellen. Dieser Workflow kombiniert Analysefunktionen in BigQuery mit dem asynchronen und skalierbaren globalen Messaging-Dienst in Pub/Sub.

Wenn Sie Daten nach Pub/Sub exportieren möchten, verwenden Sie die Anweisung EXPORT DATA in einer kontinuierlichen Abfrage, um Daten aus BigQuery in ein Pub/Sub-Thema zu exportieren.

Vorbereitung

Sie müssen ein Dienstkonto erstellen. Ein Dienstkonto ist erforderlich, um eine kontinuierliche Abfrage auszuführen, die Ergebnisse in ein Pub/Sub-Thema exportiert.

Sie müssen ein Pub/Sub-Thema erstellen, um die Ergebnisse der kontinuierlichen Abfrage als Nachrichten zu empfangen, und ein Pub/Sub-Abo, über das die Zielanwendung diese Nachrichten empfangen kann.

Erforderliche Rollen

Dieser Abschnitt enthält Informationen zu den Rollen und Berechtigungen, die für das Nutzerkonto erforderlich sind, mit dem die kontinuierliche Abfrage erstellt wird, sowie zum Dienstkonto, das die kontinuierliche Abfrage ausführt.

Nutzerkontoberechtigungen

Zum Erstellen eines Jobs in BigQuery muss das Nutzerkonto die IAM-Berechtigung bigquery.jobs.create haben. Jede der folgenden IAM-Rollen gewährt die Berechtigung bigquery.jobs.create:

• BigQuery-Nutzer (roles/bigquery.user)
• BigQuery-Jobnutzer(roles/bigquery.jobUser)
• BigQuery Admin (roles/bigquery.admin)

Wenn Sie einen Job einreichen möchten, der mit einem Dienstkonto ausgeführt wird, muss das Nutzerkonto die Rolle Dienstkontonutzer (roles/iam.serviceAccountUser) haben. Wenn Sie dasselbe Nutzerkonto zum Erstellen des Dienstkontos verwenden, muss das Nutzerkonto die Rolle Service Account Admin (roles/iam.serviceAccountAdmin) haben. Informationen dazu, wie Sie den Zugriff eines Nutzers auf ein einzelnes Dienstkonto anstelle aller Dienstkonten in einem Projekt beschränken, finden Sie unter Einzelne Rolle zuweisen.

Wenn das Nutzerkonto die APIs aktivieren muss, die für Ihren Anwendungsfall mit kontinuierlichen Abfragen erforderlich sind, muss das Nutzerkonto die Rolle Service Usage Admin (roles/serviceusage.serviceUsageAdmin) haben.

Dienstkontoberechtigungen

Zum Exportieren von Daten aus einer BigQuery-Tabelle muss das Dienstkonto die IAM-Berechtigung bigquery.tables.export haben. Jede der folgenden IAM-Rollen gewährt die Berechtigung bigquery.tables.export:

• BigQuery Data Viewer (roles/bigquery.dataViewer)
• BigQuery Data Editor (roles/bigquery.dataEditor)
• BigQuery Data Owner (roles/bigquery.dataOwner)
• BigQuery Admin (roles/bigquery.admin)

Damit das Dienstkonto auf Pub/Sub zugreifen kann, müssen Sie ihm die beiden folgenden IAM-Rollen zuweisen:

• Pub/Sub Viewer (roles/pubsub.viewer)
• Pub/Sub Publisher (roles/pubsub.publisher)

Sie können die erforderlichen Berechtigungen auch über benutzerdefinierte Rollen erhalten.

Hinweis

Aktivieren Sie die BigQuery API und die Pub/Sub API, falls sie noch nicht aktiviert sind.

Erforderliche Rollen zum Aktivieren von APIs

Zum Aktivieren von APIs benötigen Sie die Berechtigung serviceusage.services.enable. Wenn Sie das Projekt erstellt haben, haben Sie diese Berechtigung wahrscheinlich bereits über die Inhaberrolle (roles/owner). Andernfalls können Sie diese Berechtigung über die Rolle „Service Usage-Administrator“ (roles/serviceusage.serviceUsageAdmin) erhalten. Informationen zum Zuweisen von Rollen

APIs aktivieren

Nach Pub/Sub exportieren

Verwenden Sie die Anweisung EXPORT DATA, um Daten in ein Pub/Sub-Thema zu exportieren:

Console

1. Rufen Sie in der Google Cloud Console die Seite BigQuery auf.

   BigQuery aufrufen

2. Klicken Sie im Abfrageeditor auf Bearbeiten > Abfrageeinstellungen.

3. Klicken Sie im Bereich Kontinuierliche Abfrage das Kästchen Modus für kontinuierliche Abfragen verwenden an.

4. Wählen Sie im Feld Dienstkonto das erstellte Dienstkonto aus.

5. Klicken Sie auf Speichern.

6. Geben Sie im Abfrageeditor die folgende Anweisung ein:

   EXPORT DATA
   OPTIONS (
   format = 'CLOUD_PUBSUB',
   uri = 'https://pubsub.googleapis.com/projects/PROJECT_ID/topics/TOPIC_ID'
   ) AS
   (
   QUERY
   );

   Dabei gilt:

   • PROJECT_ID: Ihre Projekt-ID.
   • TOPIC_ID: die Pub/Sub-Themen-ID. Sie finden die Themen-ID auf der Seite Themen der Google Cloud Console.
   • QUERY: die SQL-Anweisung zum Auswählen der zu exportierenden Daten. Die SQL-Anweisung darf nur unterstützte Vorgänge enthalten. Sie müssen die Funktion APPENDS in der FROM-Klausel einer kontinuierlichen Abfrage verwenden, um den Zeitpunkt anzugeben, an dem die Datenverarbeitung beginnen soll.

7. Klicken Sie auf Ausführen.

bq

1. Aktivieren Sie Cloud Shell in der Google Cloud Console.

   Cloud Shell aktivieren

   Unten in der Google Cloud Console wird eine Cloud Shell-Sitzung gestartet und eine Eingabeaufforderung angezeigt. Cloud Shell ist eine Shell-Umgebung, in der das Google Cloud CLI bereits installiert ist und Werte für Ihr aktuelles Projekt bereits festgelegt sind. Das Initialisieren der Sitzung kann einige Sekunden dauern.

2. Führen Sie die kontinuierliche Abfrage in der Befehlszeile mit dem Befehl bq query und den folgenden Flags aus:

   • Legen Sie das Flag --continuous auf true fest, um die Abfrage kontinuierlich auszuführen.
   • Verwenden Sie das Flag --connection_property, um ein Dienstkonto anzugeben.

   bq query --project_id=PROJECT_ID --use_legacy_sql=false \
   --continuous=true --connection_property=service_account=SERVICE_ACCOUNT_EMAIL \
   'EXPORT DATA OPTIONS (format = "CLOUD_PUBSUB", uri = "https://pubsub.googleapis.com/projects/PROJECT_ID/topics/TOPIC_ID") AS (QUERY);'

   Ersetzen Sie Folgendes:

   • PROJECT_ID: Ihre Projekt-ID.
   • SERVICE_ACCOUNT_EMAIL: die E-Mail-Adresse des Dienstkontos. Die E‑Mail-Adresse des Dienstkontos finden Sie in der Google Cloud Console auf der Seite Dienstkonten.
   • QUERY: die SQL-Anweisung zum Auswählen der zu exportierenden Daten. Die SQL-Anweisung darf nur unterstützte Vorgänge enthalten. Sie müssen die Funktion APPENDS in der FROM-Klausel einer kontinuierlichen Abfrage verwenden, um den Zeitpunkt anzugeben, an dem die Datenverarbeitung beginnen soll.

API

1. Führen Sie die kontinuierliche Abfrage aus, indem Sie die Methode jobs.insert aufrufen. Legen Sie die folgenden Felder in der JobConfigurationQuery-Ressource der Job-Ressource fest, die Sie übergeben:

   • Legen Sie das Feld continuous auf true fest, um die Abfrage kontinuierlich zu machen.
   • Verwenden Sie das Feld connection_property, um ein Dienstkonto anzugeben.

   curl --request POST \
     'https://bigquery.googleapis.com/bigquery/v2/projects/PROJECT_ID/jobs'
     --header 'Authorization: Bearer $(gcloud auth print-access-token) \
     --header 'Accept: application/json' \
     --header 'Content-Type: application/json' \
     --data '("configuration":("query":"EXPORT DATA OPTIONS (format = 'CLOUD_PUBSUB', uri = 'https://pubsub.googleapis.com/projects/PROJECT_ID/topics/TOPIC_ID') AS (QUERY);","useLegacySql":false,"continuous":true,"connectionProperties":["key": "service_account","value":"SERVICE_ACCOUNT_EMAIL"]))' \
     --compressed

   Ersetzen Sie Folgendes:

   • PROJECT_ID: Ihre Projekt-ID.
   • QUERY: die SQL-Anweisung zum Auswählen der zu exportierenden Daten. Die SQL-Anweisung darf nur unterstützte Vorgänge enthalten. Sie müssen die Funktion APPENDS in der FROM-Klausel einer kontinuierlichen Abfrage verwenden, um den Zeitpunkt anzugeben, an dem die Datenverarbeitung beginnen soll.
   • SERVICE_ACCOUNT_EMAIL: die E-Mail-Adresse des Dienstkontos. Die E‑Mail-Adresse des Dienstkontos finden Sie in der Google Cloud Console auf der Seite Dienstkonten.

Mehrere Spalten nach Pub/Sub exportieren

Wenn Sie mehrere Spalten in die Ausgabe aufnehmen möchten, können Sie eine Strukturspalte erstellen, die die Spaltenwerte enthält, und den Strukturwert dann mithilfe der TO_JSON_STRING-Funktion in einen JSON-Strong konvertieren. Im folgenden Beispiel werden Daten aus vier Spalten exportiert, die als JSON-String formatiert sind:

EXPORT DATA
  OPTIONS (
    format = 'CLOUD_PUBSUB',
    uri = 'https://pubsub.googleapis.com/projects/myproject/topics/taxi-real-time-rides')
AS (
  SELECT
    TO_JSON_STRING(
      STRUCT(
        ride_id,
        timestamp,
        latitude,
        longitude)) AS message
  FROM
    APPENDS(TABLE `myproject.real_time_taxi_streaming.taxi_rides`,
      -- Configure the APPENDS TVF start_timestamp to specify when you want to
      -- start processing data using your continuous query.
      -- This example starts processing at 10 minutes before the current time.
      CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  WHERE ride_status = 'enroute'
);

Exportoptimierung

Wenn die Leistung Ihres kontinuierlichen Abfragejobs durch die verfügbaren Rechenressourcen eingeschränkt zu sein scheint, versuchen Sie, die Größe der BigQuery-CONTINUOUS-Slot-Reservierungszuweisung zu erhöhen.

Beschränkungen

• Die exportierten Daten müssen aus einer einzelnen STRING- oder BYTES-Spalte bestehen. Der Spaltenname kann beliebig gewählt werden.
• Für den Export nach Pub/Sub müssen Sie eine kontinuierliche Abfrage verwenden.
• Sie können kein Schema an ein Pub/Sub-Thema in der kontinuierlichen Abfrage übergeben.
• Sie können keine Daten in ein Pub/Sub-Thema exportieren, das ein Schema verwendet.
• Beim Exportieren nach Pub/Sub können Sie JSON-formatierte Datensätze exportieren, bei denen einige Werte NULL sind. Sie können jedoch keine Datensätze exportieren, die nur aus NULL-Werten bestehen. Sie können NULL-Einträge aus den Abfrageergebnissen ausschließen, indem Sie einen WHERE message IS NOT NULL-Filter in die kontinuierliche Abfrage einfügen.
• Wenn Sie Daten in ein Pub/Sub-Thema exportieren, das mit einem Standortendpunkt konfiguriert ist, muss der Endpunkt innerhalb derselben regionalen Grenze von Google Cloud konfiguriert sein wie das BigQuery-Dataset, das die Tabelle enthält, die Sie abfragen.
• Exportierte Daten dürfen die Pub/Sub-Kontingente nicht überschreiten.

Preise

Wenn Sie Daten in einer kontinuierlichen Abfrage exportieren, werden Ihnen die Kosten gemäß den BigQuery-Kapazitätsrechenpreisen in Rechnung gestellt. Zum Ausführen kontinuierlicher Abfragen benötigen Sie eine Reservierung mit dem Enterprise oder Enterprise Plus und eine Reservierungszuweisung mit dem CONTINUOUS-Jobtyp.

Nachdem die Daten exportiert wurden, wird die Verwendung von Pub/Sub in Rechnung gestellt. Weitere Informationen finden Sie unter Preise für Cloud Pub/Sub.