Empfangene Ereignisse formatieren

Eine Pipeline verbindet einen Bus mit einem Ziel und leitet Ereignisnachrichten an dieses Ziel weiter. Sie können eine Pipeline so konfigurieren, dass sie Ereignisdaten in einem bestimmten Format erwartet. Alternativ können Sie Ereignisdaten von einem unterstützten Format in ein anderes konvertieren, bevor Ereignisse an ein Ziel gesendet werden. Möglicherweise müssen Sie Ereignisse beispielsweise an einen Endpunkt weiterleiten, der nur Avro-Daten akzeptiert.

Unterstützte Formate

Die folgenden Formatkonvertierungen werden unterstützt:

  • Avro zu JSON
  • Avro zu Protobuf
  • JSON zu Avro
  • JSON zu Protobuf
  • Protobuf zu Avro
  • Protobuf zu JSON

Wichtige Hinweise:

  • Wenn Sie das Format von Ereignissen konvertieren, wird nur die Ereignisnutzlast und nicht die gesamte Ereignisnachricht konvertiert.

  • Wenn für eine Pipeline ein Eingangsdatenformat angegeben ist, müssen alle Ereignisse diesem Format entsprechen. Alle Ereignisse, die nicht dem erwarteten Format entsprechen, werden als dauerhafte Fehler behandelt.

  • Wenn für eine Pipeline kein Eingangsdatenformat angegeben ist, kann kein Ausgangsformat festgelegt werden.

  • Bevor ein Ereignisformat für ein bestimmtes Ziel konvertiert wird, wird zuerst jede Datentransformation angewendet, die konfiguriert ist.

  • Ereignisse werden immer in einem CloudEvents-Format mit einer HTTP-Anfrage im Binärinhaltsmodus gesendet , es sei denn, Sie geben eine Nachrichtenbindung an.

  • JSON-Schemas werden dynamisch erkannt. Für Protobuf-Schemadefinitionen können Sie nur einen Typ der obersten Ebene definieren. Importanweisungen, die auf andere Typen verweisen, werden nicht unterstützt. Bei Schemadefinitionen ohne syntax-Kennung wird standardmäßig proto2 verwendet. Beachten Sie, dass die Schemagröße begrenzt ist.

Pipeline zum Formatieren von Ereignissen konfigurieren

Sie können eine Pipeline so konfigurieren, dass sie Ereignisdaten in einem bestimmten Format erwartet, oder um Ereignisdaten in der Google Cloud Konsole oder durch Verwendung der gcloud CLI von einem Format in ein anderes zu konvertieren.

Console

  1. Rufen Sie in der Google Cloud Console die Seite Eventarc > Pipelines auf.

    Zu Pipelines

  2. Sie können eine Pipeline erstellen oder, wenn Sie eine Pipeline aktualisieren, auf den Namen der Pipeline klicken.

  3. Klicken Sie auf der Seite Pipelines-Details auf Bearbeiten.

  4. Gehen Sie im Bereich Ereignisvermittlung so vor:

    1. Wählen Sie das Kästchen Transformation anwenden aus.

    2. Wählen Sie in der Liste Eingangsformat das entsprechende Format aus.

      Wenn für eine Pipeline ein Eingangsdatenformat angegeben ist, müssen alle Ereignisse diesem Format entsprechen. Alle Ereignisse, die nicht dem erwarteten Format entsprechen, werden als dauerhafte Fehler behandelt.

    3. Für Avro- oder Protobuf-Formate müssen Sie ein Eingangsschema angeben. Optional können Sie ein Eingangsschema hochladen, anstatt es direkt anzugeben.

    4. Schreiben Sie im Feld CEL-Ausdruck einen Transformationsausdruck mit CEL.

    5. Klicken Sie auf Weiter.

  5. Gehen Sie im Bereich Ziel so vor:

    1. Wählen Sie gegebenenfalls in der Liste Ausgangsformat ein Format aus.

      Wenn für eine Pipeline kein Eingangsdatenformat angegeben ist, kann kein Ausgangsformat festgelegt werden.

    2. Optional: Wenden Sie eine Nachrichtenbindung an. Weitere Informationen finden Sie unter Nachrichtenbindung.

  6. Klicken Sie auf Speichern.

    Es kann einige Minuten dauern, bis eine Pipeline aktualisiert ist.

gcloud

  1. Öffnen Sie ein Terminalfenster.

  2. Sie können eine Pipeline erstellen oder eine Pipeline mit dem gcloud eventarc pipelines update Befehl aktualisieren:

    gcloud eventarc pipelines update PIPELINE_NAME \
    --location=REGION \
    --INPUT_PAYLOAD_FLAG \
    --destinations=OUTPUT_PAYLOAD_KEY

    Ersetzen Sie Folgendes:

    • PIPELINE_NAME: die ID der Pipeline oder ein voll qualifizierter Name

    • REGION: ein unterstützter Eventarc Advanced-Standort

      Alternativ können Sie das Standortattribut der gcloud CLI festlegen:

      gcloud config set eventarc/location REGION

    • INPUT_PAYLOAD_FLAG: ein Flag für das Eingangsdatenformat, das einen der folgenden Werte haben kann:

      • --input-payload-format-avro-schema-definition
      • --input-payload-format-json
      • --input-payload-format-protobuf-schema-definition

      Beachten Sie, dass, wenn für eine Pipeline ein Eingangsdatenformat angegeben ist, alle Ereignisse diesem Format entsprechen müssen. Alle Ereignisse, die nicht dem erwarteten Format entsprechen, werden als dauerhafte Fehler behandelt.

    • OUTPUT_PAYLOAD_KEY: ein Schlüssel für das Ausgangsdatenformat, der einen der folgenden Werte haben kann:

      • output_payload_format_avro_schema_definition
      • output_payload_format_json
      • output_payload_format_protobuf_schema_definition

      Beachten Sie, dass Sie, wenn Sie einen Schlüssel für das Ausgabedatenformat festlegen, auch ein Flag für das Eingabedatenformat angeben müssen.

    Es kann einige Minuten dauern, bis eine Pipeline aktualisiert ist.

    Beispiele:

    Im folgenden Beispiel wird mit dem Flag --input-payload-format-protobuf-schema-definition angegeben, dass die Pipeline Ereignisse im Protobuf-Datenformat mit einem bestimmten Schema erwarten soll:

    gcloud eventarc pipelines update my-pipeline \
    --input-payload-format-protobuf-schema-definition \
'
  syntax = "proto3";
  message schema {
    string name = 1;
    string severity = 2;
  }
'

    Im folgenden Beispiel werden ein output_payload_format_avro_schema_definition-Schlüssel und ein --input-payload-format-avro-schema-definition-Flag verwendet, um eine Pipeline zu erstellen, die Ereignisse im Avro-Format erwartet und sie im selben Format ausgibt:

    gcloud eventarc pipelines create my-pipeline \
    --location=us-central1 \
    --destinations=http_endpoint_uri='https://example-endpoint.com',output_payload_format_avro_schema_definition='{"type": "record", "name": "my_record", "fields": [{"name": "my_field", "type": "string"}]}' \
    --input-payload-format-avro-schema-definition='{"type": "record", "name": "my_record", "fields": [{"name": "my_field", "type": "string"}]}'

    Im folgenden Beispiel werden ein output_payload_format_protobuf_schema_definition-Schlüssel und ein --input-payload-format-avro-schema-definition-Flag verwendet, um eine Pipeline zu aktualisieren und ihre Ereignisdaten mithilfe von Schemadefinitionen von Avro in Protobuf zu konvertieren:

    gcloud eventarc pipelines update my-pipeline \
    --location=us-central1 \
    --destinations=output_payload_format_protobuf_schema_definition='message MessageProto {string prop1 = 1; string prop2 = 2;}' \
    --input-payload-format-avro-schema-definition= \
    '
    {
      "type": "record",
      "name": "MessageProto",
      "fields": [
        { "name" : "prop1", "type": "string" },
        { "name" : "prop2", "type": "string" },
      ]
    }
    '