Aktywowanie DAG-a w Node.JS i Google Cloud Functions

1. Wprowadzenie

Usługa Apache Airflow została zaprojektowana tak, aby regularnie uruchamiać DAG-i, ale możesz też aktywować je w odpowiedzi na zdarzenia, na przykład zmianę w zasobniku Cloud Storage lub wiadomość przesłaną do Cloud Pub/Sub. W tym celu DAG-i w usłudze Cloud Composer mogą być aktywowane przez Cloud Functions.

W tym module przykładowy DAG jest uruchamiany za każdym razem, gdy w zasobniku Cloud Storage wystąpi zmiana. Ten DAG używa operatora BashOperator do uruchomienia polecenia bash, w którym wyświetlane są informacje o zmianie dotyczące danych przesłanych do zasobnika Cloud Storage.

Przed rozpoczęciem tego modułu zalecamy ukończenie modułów z programowania Wprowadzenie do Cloud Composer i Pierwsze kroki z Cloud Functions. Jeśli w ramach szkolenia wprowadzającego do Cloud Composer utworzysz środowisko Composer, możesz go użyć w tym module.

Co zbudujesz

W ramach tego ćwiczenia w Codelabs:

  1. Przesłać plik do Google Cloud Storage, co spowoduje
  2. Aktywuj funkcję w Google Cloud Functions za pomocą środowiska wykonawczego Node.JS
  3. Ta funkcja uruchomi DAG-a w usłudze Google Cloud Composer
  4. Powoduje to uruchomienie prostego polecenia bash wyświetlającego zmianę w zasobniku Google Cloud Storage

1d3d3736624a923f.png

Czego się nauczysz

  • Jak aktywować DAG-a Apache Airflow za pomocą Google Cloud Functions + Node.js

Potrzebne elementy

  • Konto GCP
  • Podstawowa znajomość języka JavaScript
  • Podstawowa znajomość usługi Cloud Composer/Airflow i Cloud Functions
  • Komfort korzystania z poleceń interfejsu wiersza poleceń

2. Konfigurowanie GCP

Wybierz lub utwórz projekt

Wybierz lub utwórz projekt Google Cloud Platform. Jeśli tworzysz nowy projekt, wykonaj czynności opisane tutaj.

Zapisz identyfikator projektu. Użyjesz go w kolejnych krokach.

Jeśli tworzysz nowy projekt, jego identyfikator znajduje się tuż pod jego nazwą na stronie tworzenia.

Jeśli masz już utworzony projekt, identyfikator znajdziesz na stronie głównej konsoli na karcie Informacje o projekcie.

Włączanie interfejsów API

Włącz Cloud Composer, Google Cloud Functions i Cloud Identity oraz interfejs Google Identity and Access Management (IAM) API.

Tworzenie środowiska Composer

Utwórz środowisko Cloud Composer o tej konfiguracji:

  • Nazwa: my-composer-environment
  • Lokalizacja: dowolna lokalizacja, która znajduje się najbliżej Ciebie.
  • Strefa: dowolna strefa w danym regionie.

Wszystkie inne konfiguracje mogą pozostać domyślne. Kliknij „Utwórz”. na dole.Zapisz nazwę i lokalizację środowiska Composer – będą potrzebne w kolejnych krokach.

Tworzenie zasobnika Cloud Storage

Utwórz w projekcie zasobnik Cloud Storage o tej konfiguracji:

  • Nazwa: <identyfikator-twojego-projektu>
  • Domyślna klasa pamięci masowej: wiele regionów
  • Lokalizacja: dowolna lokalizacja znajdująca się najbliżej używanego przez Ciebie regionu Cloud Composer
  • Model kontroli dostępu: ustawianie uprawnień na poziomie obiektu i zasobnika

Kliknij „Utwórz”. Gdy wszystko będzie gotowe, zapisz nazwę swojego zasobnika Cloud Storage – do wykorzystania w kolejnych krokach.

3. Konfigurowanie Google Cloud Functions (GCF)

Aby skonfigurować GCF, uruchomimy polecenia w Google Cloud Shell.

Google Cloud można obsługiwać zdalnie z poziomu laptopa za pomocą narzędzia wiersza poleceń gcloud, ale w tym ćwiczeniu z programowania wykorzystamy Google Cloud Shell – środowisko wiersza poleceń działające w chmurze.

Ta maszyna wirtualna oparta na Debianie zawiera wszystkie potrzebne narzędzia dla programistów. Zawiera stały katalog domowy o pojemności 5 GB i działa w Google Cloud, znacząco zwiększając wydajność sieci i uwierzytelnianie. Oznacza to, że do tego ćwiczenia z programowania wystarczy przeglądarka (tak, działa ona na Chromebooku).

Aby aktywować Google Cloud Shell, w konsoli programisty kliknij przycisk w prawym górnym rogu (udostępnienie środowiska i połączenie z nim powinno zająć tylko kilka chwil):

Przyznawanie uprawnień do podpisywania obiektów blob kontu usługi Cloud Functions

Aby GCF mogło uwierzytelnić się w Cloud IAP, czyli serwerze proxy chroniącym serwer WWW Airflow, musisz przypisać do konta usługi Appspot rolę Service Account Token Creator w GCF. Aby to zrobić, uruchom w Cloud Shell podane niżej polecenie, zastępując nazwę projektu nazwą <your-project-id>.

gcloud iam service-accounts add-iam-policy-binding \
<your-project-id>@appspot.gserviceaccount.com \
--member=serviceAccount:<your-project-id>@appspot.gserviceaccount.com \
--role=roles/iam.serviceAccountTokenCreator

Jeśli na przykład projekt nazywa się my-project, polecenie będzie wyglądać tak:

gcloud iam service-accounts add-iam-policy-binding \
my-project@appspot.gserviceaccount.com \
--member=serviceAccount:my-project@appspot.gserviceaccount.com \
--role=roles/iam.serviceAccountTokenCreator

Uzyskiwanie identyfikatora klienta

Aby utworzyć token na potrzeby uwierzytelniania w Cloud IAP, funkcja wymaga identyfikatora klienta serwera proxy, który chroni serwer WWW Airflow. Interfejs Cloud Composer API nie udostępnia tych informacji bezpośrednio. Zamiast tego wyślij nieuwierzytelnione żądanie do serwera WWW Airflow i przechwyć identyfikator klienta z adresu URL przekierowania. W tym celu uruchomimy plik Pythona za pomocą Cloud Shell, aby przechwycić identyfikator klienta.

Pobierz niezbędny kod z GitHuba, uruchamiając w Cloud Shell to polecenie:

cd
git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git

Jeśli wystąpił błąd, ponieważ ten katalog już istnieje, zaktualizuj go do najnowszej wersji, uruchamiając następujące polecenie

cd python-docs-samples/
git pull origin master

Przejdź do odpowiedniego katalogu, uruchamiając polecenie

cd python-docs-samples/composer/rest

Uruchom kod Pythona, aby uzyskać identyfikator klienta, zastępując nazwę projektu <your-project-id>, lokalizację utworzonego wcześniej środowiska Composer dla usługi <your-composer-location> oraz nazwę utworzonego wcześniej środowiska Composer dla platformy <your-composer-environment>

python3 get_client_id.py <your-project-id> <your-composer-location> <your-composer-environment>

Jeśli na przykład nazwa projektu to my-project, lokalizacja w usłudze Composer to us-central1, a nazwa środowiska to my-composer, polecenie będzie wyglądać tak:

python3 get_client_id.py my-project us-central1 my-composer

get_client_id.py wykonuje te czynności:

  • Uwierzytelnia się w Google Cloud
  • Wykonuje nieuwierzytelnione żądanie HTTP do serwera WWW Airflow, aby uzyskać identyfikator URI przekierowania
  • Wyodrębnia z przekierowania parametr zapytania client_id
  • Drukuje go do użycia

Twój identyfikator klienta zostanie wyświetlony w wierszu poleceń i będzie wyglądał mniej więcej tak:

12345678987654321-abc1def3ghi5jkl7mno8pqr0.apps.googleusercontent.com

4. Utwórz funkcję

Skopiuj w Cloud Shell repozytorium z odpowiednim przykładowym kodem, uruchamiając polecenie

cd
git clone https://github.com/GoogleCloudPlatform/nodejs-docs-samples.git

Przejdź do niezbędnego katalogu i nie zamykaj Cloud Shell podczas wykonywania kolejnych kroków.

cd nodejs-docs-samples/composer/functions/composer-storage-trigger

Otwórz stronę Google Cloud Functions – w tym celu kliknij menu nawigacyjne, a następnie „Cloud Functions”.

Kliknij „UTWÓRZ FUNKCJĘ” na górze strony.

Nazwij funkcję „moja-funkcja”. i zostaw domyślną wartość 256 MB.

Ustaw aktywator na „Cloud Storage”, pozostaw typ zdarzenia „Finalizuj/Utwórz” i przejdź do zasobnika utworzonego w kroku Utwórz zasobnik Cloud Storage.

W polu Kod źródłowy pozostaw ustawienie „Wbudowany edytor”. i ustaw środowisko wykonawcze na „Node.js 8”

W Cloud Shell uruchom podane niżej polecenie. Spowoduje to otwarcie pliku index.js i pakietu package.json w edytorze Cloud Shell.

cloudshell edit index.js package.json

Kliknij kartę package.json, skopiuj ten kod i wklej go do sekcji package.json we wbudowanym edytorze Cloud Functions

Ustaw funkcję do wykonania. aby aktywowaćDag

Kliknij kartę index.js, skopiuj kod i wklej go w sekcji index.js we wbudowanym edytorze Cloud Functions

Zmień PROJECT_ID na identyfikator projektu, a CLIENT_ID na identyfikator klienta zapisany w kroku „Uzyskiwanie identyfikatora klienta”. NIE klikaj „Utwórz” jeszcze jednak musisz wypełnić jeszcze kilka innych pól!

W Cloud Shell uruchom podane niżej polecenie, zastępując fragment <nazwa-środowiska> nazwą środowiska Composer i region <Twój-kompozytor-kompozytor> z regionem, w którym znajduje się środowisko Composer.

gcloud composer environments describe <your-environment-name> --location <your-composer-region>

Jeśli na przykład środowisko ma nazwę my-composer-environment i znajduje się w regionie us-central1, polecenie będzie wyglądać tak:

gcloud composer environments describe my-composer-environment --location us-central1

Dane wyjściowe powinny wyglądać mniej więcej tak:

config:
 airflowUri: https://abc123efghi456k-tp.appspot.com
 dagGcsPrefix: gs://narnia-north1-test-codelab-jklmno-bucket/dags
 gkeCluster: projects/a-project/zones/narnia-north1-b/clusters/narnia-north1-test-codelab-jklmno-gke
 nodeConfig:
   diskSizeGb: 100
   location: projects/a-project/zones/narnia-north1-b
   machineType: projects/a-project/zones/narnia-north1-b/machineTypes/n1-standard-1
   network: projects/a-project/global/networks/default
   oauthScopes:
   - https://www.googleapis.com/auth/cloud-platform
   serviceAccount: 987665432-compute@developer.gserviceaccount.com
 nodeCount: 3
 softwareConfig:
   imageVersion: composer-1.7.0-airflow-1.10.0
   pythonVersion: '2'
createTime: '2019-05-29T09:41:27.919Z'
name: projects/a-project/locations/narnia-north1/environments/my-composer-environment
state: RUNNING
updateTime: '2019-05-29T09:56:29.969Z'
uuid: 123456-7890-9876-543-210123456

W danych wyjściowych odszukaj zmienną o nazwie airflowUri. W kodzie index.js zmień WEBSERVER_ID na identyfikator serwera WWW Airflow – jest to część zmiennej airflowUri, która zawiera „-tp” na końcu, na przykład abc123efghi456k-tp

Kliknij przycisk „Więcej”. i wybierz najbliższy geograficznie region

Zaznacz opcję „Ponów próbę w przypadku niepowodzenia”.

Kliknij „Utwórz”. aby utworzyć funkcję w Cloud Functions

Analiza kodu

Kod skopiowany z pliku index.js będzie wyglądał mniej więcej tak:

// [START composer_trigger]
'use strict';

const fetch = require('node-fetch');
const FormData = require('form-data');

/**
 * Triggered from a message on a Cloud Storage bucket.
 *
 * IAP authorization based on:
 * https://stackoverflow.com/questions/45787676/how-to-authenticate-google-cloud-functions-for-access-to-secure-app-engine-endpo
 * and
 * https://cloud.google.com/iap/docs/authentication-howto
 *
 * @param {!Object} data The Cloud Functions event data.
 * @returns {Promise}
 */
exports.triggerDag = async data => {
  // Fill in your Composer environment information here.

  // The project that holds your function
  const PROJECT_ID = 'your-project-id';
  // Navigate to your webserver's login page and get this from the URL
  const CLIENT_ID = 'your-iap-client-id';
  // This should be part of your webserver's URL:
  // {tenant-project-id}.appspot.com
  const WEBSERVER_ID = 'your-tenant-project-id';
  // The name of the DAG you wish to trigger
  const DAG_NAME = 'composer_sample_trigger_response_dag';

  // Other constants
  const WEBSERVER_URL = `https://${WEBSERVER_ID}.appspot.com/api/experimental/dags/${DAG_NAME}/dag_runs`;
  const USER_AGENT = 'gcf-event-trigger';
  const BODY = {conf: JSON.stringify(data)};

  // Make the request
  try {
    const iap = await authorizeIap(CLIENT_ID, PROJECT_ID, USER_AGENT);

    return makeIapPostRequest(
      WEBSERVER_URL,
      BODY,
      iap.idToken,
      USER_AGENT,
      iap.jwt
    );
  } catch (err) {
    throw new Error(err);
  }
};

/**
 * @param {string} clientId The client id associated with the Composer webserver application.
 * @param {string} projectId The id for the project containing the Cloud Function.
 * @param {string} userAgent The user agent string which will be provided with the webserver request.
 */
const authorizeIap = async (clientId, projectId, userAgent) => {
  const SERVICE_ACCOUNT = `${projectId}@appspot.gserviceaccount.com`;
  const JWT_HEADER = Buffer.from(
    JSON.stringify({alg: 'RS256', typ: 'JWT'})
  ).toString('base64');

  let jwt = '';
  let jwtClaimset = '';

  // Obtain an Oauth2 access token for the appspot service account
  const res = await fetch(
    `http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/${SERVICE_ACCOUNT}/token`,
    {
      headers: {'User-Agent': userAgent, 'Metadata-Flavor': 'Google'},
    }
  );
  const tokenResponse = await res.json();
  if (tokenResponse.error) {
    return Promise.reject(tokenResponse.error);
  }

  const accessToken = tokenResponse.access_token;
  const iat = Math.floor(new Date().getTime() / 1000);
  const claims = {
    iss: SERVICE_ACCOUNT,
    aud: 'https://www.googleapis.com/oauth2/v4/token',
    iat: iat,
    exp: iat + 60,
    target_audience: clientId,
  };
  jwtClaimset = Buffer.from(JSON.stringify(claims)).toString('base64');
  const toSign = [JWT_HEADER, jwtClaimset].join('.');

  const blob = await fetch(
    `https://iam.googleapis.com/v1/projects/${projectId}/serviceAccounts/${SERVICE_ACCOUNT}:signBlob`,
    {
      method: 'POST',
      body: JSON.stringify({
        bytesToSign: Buffer.from(toSign).toString('base64'),
      }),
      headers: {
        'User-Agent': userAgent,
        Authorization: `Bearer ${accessToken}`,
      },
    }
  );
  const blobJson = await blob.json();
  if (blobJson.error) {
    return Promise.reject(blobJson.error);
  }

  // Request service account signature on header and claimset
  const jwtSignature = blobJson.signature;
  jwt = [JWT_HEADER, jwtClaimset, jwtSignature].join('.');
  const form = new FormData();
  form.append('grant_type', 'urn:ietf:params:oauth:grant-type:jwt-bearer');
  form.append('assertion', jwt);

  const token = await fetch('https://www.googleapis.com/oauth2/v4/token', {
    method: 'POST',
    body: form,
  });
  const tokenJson = await token.json();
  if (tokenJson.error) {
    return Promise.reject(tokenJson.error);
  }

  return {
    jwt: jwt,
    idToken: tokenJson.id_token,
  };
};

/**
 * @param {string} url The url that the post request targets.
 * @param {string} body The body of the post request.
 * @param {string} idToken Bearer token used to authorize the iap request.
 * @param {string} userAgent The user agent to identify the requester.
 */
const makeIapPostRequest = async (url, body, idToken, userAgent) => {
  const res = await fetch(url, {
    method: 'POST',
    headers: {
      'User-Agent': userAgent,
      Authorization: `Bearer ${idToken}`,
    },
    body: JSON.stringify(body),
  });

  if (!res.ok) {
    const err = await res.text();
    throw new Error(err);
  }
};
// [END composer_trigger]

Zobaczmy, co się dzieje. Są tu 3 funkcje: triggerDag, authorizeIap i makeIapPostRequest

triggerDag to funkcja aktywowana po przesłaniu czegoś do wyznaczonego zasobnika Cloud Storage. To tutaj konfigurujemy ważne zmienne używane w innych żądaniach, np. PROJECT_ID, CLIENT_ID, WEBSERVER_ID i DAG_NAME. Wywołuje on połączenia authorizeIap i makeIapPostRequest.

exports.triggerDag = async data => {
  // Fill in your Composer environment information here.

  // The project that holds your function
  const PROJECT_ID = 'your-project-id';
  // Navigate to your webserver's login page and get this from the URL
  const CLIENT_ID = 'your-iap-client-id';
  // This should be part of your webserver's URL:
  // {tenant-project-id}.appspot.com
  const WEBSERVER_ID = 'your-tenant-project-id';
  // The name of the DAG you wish to trigger
  const DAG_NAME = 'composer_sample_trigger_response_dag';

  // Other constants
  const WEBSERVER_URL = `https://${WEBSERVER_ID}.appspot.com/api/experimental/dags/${DAG_NAME}/dag_runs`;
  const USER_AGENT = 'gcf-event-trigger';
  const BODY = {conf: JSON.stringify(data)};

  // Make the request
  try {
    const iap = await authorizeIap(CLIENT_ID, PROJECT_ID, USER_AGENT);

    return makeIapPostRequest(
      WEBSERVER_URL,
      BODY,
      iap.idToken,
      USER_AGENT,
      iap.jwt
    );
  } catch (err) {
    throw new Error(err);
  }
};

authorizeIap wysyła żądanie do serwera proxy, który chroni serwer internetowy Airflow przy użyciu konta usługi i funkcji wymiany JWT dla tokena tożsamości, który będzie używany do uwierzytelniania makeIapPostRequest.

const authorizeIap = async (clientId, projectId, userAgent) => {
  const SERVICE_ACCOUNT = `${projectId}@appspot.gserviceaccount.com`;
  const JWT_HEADER = Buffer.from(
    JSON.stringify({alg: 'RS256', typ: 'JWT'})
  ).toString('base64');

  let jwt = '';
  let jwtClaimset = '';

  // Obtain an Oauth2 access token for the appspot service account
  const res = await fetch(
    `http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/${SERVICE_ACCOUNT}/token`,
    {
      headers: {'User-Agent': userAgent, 'Metadata-Flavor': 'Google'},
    }
  );
  const tokenResponse = await res.json();
  if (tokenResponse.error) {
    return Promise.reject(tokenResponse.error);
  }

  const accessToken = tokenResponse.access_token;
  const iat = Math.floor(new Date().getTime() / 1000);
  const claims = {
    iss: SERVICE_ACCOUNT,
    aud: 'https://www.googleapis.com/oauth2/v4/token',
    iat: iat,
    exp: iat + 60,
    target_audience: clientId,
  };
  jwtClaimset = Buffer.from(JSON.stringify(claims)).toString('base64');
  const toSign = [JWT_HEADER, jwtClaimset].join('.');

  const blob = await fetch(
    `https://iam.googleapis.com/v1/projects/${projectId}/serviceAccounts/${SERVICE_ACCOUNT}:signBlob`,
    {
      method: 'POST',
      body: JSON.stringify({
        bytesToSign: Buffer.from(toSign).toString('base64'),
      }),
      headers: {
        'User-Agent': userAgent,
        Authorization: `Bearer ${accessToken}`,
      },
    }
  );
  const blobJson = await blob.json();
  if (blobJson.error) {
    return Promise.reject(blobJson.error);
  }

  // Request service account signature on header and claimset
  const jwtSignature = blobJson.signature;
  jwt = [JWT_HEADER, jwtClaimset, jwtSignature].join('.');
  const form = new FormData();
  form.append('grant_type', 'urn:ietf:params:oauth:grant-type:jwt-bearer');
  form.append('assertion', jwt);

  const token = await fetch('https://www.googleapis.com/oauth2/v4/token', {
    method: 'POST',
    body: form,
  });
  const tokenJson = await token.json();
  if (tokenJson.error) {
    return Promise.reject(tokenJson.error);
  }

  return {
    jwt: jwt,
    idToken: tokenJson.id_token,
  };
};

makeIapPostRequest wywołuje serwer internetowy Airflow, aby aktywować composer_sample_trigger_response_dag.. Nazwa DAG-a jest umieszczona w adresie URL serwera WWW Airflow przekazywanym z parametrem url, a idToken to token uzyskany w żądaniu authorizeIap.

const makeIapPostRequest = async (url, body, idToken, userAgent) => {
  const res = await fetch(url, {
    method: 'POST',
    headers: {
      'User-Agent': userAgent,
      Authorization: `Bearer ${idToken}`,
    },
    body: JSON.stringify(body),
  });

  if (!res.ok) {
    const err = await res.text();
    throw new Error(err);
  }
};

5. Konfigurowanie DAG-a

W Cloud Shell przejdź do katalogu z przykładowymi przepływami pracy. Jest to część próbki pliku Python-docs, która została pobrana z GitHuba na etapie pobierania identyfikatora klienta.

cd
cd python-docs-samples/composer/workflows

Przesyłanie DAG-a do usługi Composer

Prześlij przykładowy DAG do zasobnika na dane DAG-a w środowisku Composer za pomocą tego polecenia, używając tego polecenia. <environment_name> to nazwa środowiska Composer, a <location> to nazwa regionu, w którym się znajduje. trigger_response_dag.py to DAG, nad którym będziemy pracować.

gcloud composer environments storage dags import \
    --environment <environment_name> \
    --location <location> \
    --source trigger_response_dag.py

Jeśli na przykład środowisko Composer miało nazwę my-composer i znajdowało się w regionie us-central1, polecenie będzie wyglądać tak:

gcloud composer environments storage dags import \
    --environment my-composer \
    --location us-central1 \
    --source trigger_response_dag.py

Omówienie DAG-a

Kod DAG-a w trigger_response.py wygląda tak

import datetime
import airflow
from airflow.operators import bash_operator


default_args = {
    'owner': 'Composer Example',
    'depends_on_past': False,
    'email': [''],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'start_date': datetime.datetime(2017, 1, 1),
}

with airflow.DAG(
        'composer_sample_trigger_response_dag',
        default_args=default_args,
        # Not scheduled, trigger only
        schedule_interval=None) as dag:

    # Print the dag_run's configuration, which includes information about the
    # Cloud Storage object change.
    print_gcs_info = bash_operator.BashOperator(
        task_id='print_gcs_info', bash_command='echo {{ dag_run.conf }}')

Sekcja default_args zawiera argumenty domyślne wymagane przez model BaseOperator w Apache Airflow. Zobaczysz tę sekcję z tymi parametrami w każdym DAG-u Apache Airflow. owner jest obecnie ustawiony na Composer Example, ale jeśli chcesz, możesz zmienić go na swoje imię. depends_on_past pokazuje, że ten DAG nie jest zależny od żadnych wcześniejszych. Trzy sekcje e-maili (email, email_on_failure i email_on_retry) są skonfigurowane tak, że na podstawie stanu tego DAG-a nie są wysyłane żadne e-maile z powiadomieniami. DAG spróbuje ponowić próbę tylko raz, ponieważ parametr retries ma wartość 1 (po 5 minutach na retry_delay). W połączeniu z ustawieniem schedule_interval (ustawionym później) start_date określa zwykle, kiedy DAG-a ma się uruchamiać, ale w przypadku tego DAG-a jest nieistotny. Ustawiono datę 1 stycznia 2017 r., ale można ustawić ją na dowolną z przeszłości.

default_args = {
    'owner': 'Composer Example',
    'depends_on_past': False,
    'email': [''],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'start_date': datetime.datetime(2017, 1, 1),
}

Sekcja with airflow.DAG konfiguruje DAG-a, który zostanie uruchomiony. Zostanie ono uruchomione z identyfikatorem zadania composer_sample_trigger_response_dag (domyślnymi argumentami z sekcji default_args) i – co najważniejsze – z ustawieniem schedule_interval o wartości None. schedule_interval ma wartość None, ponieważ wyzwalamy ten konkretny DAG za pomocą funkcji w Cloud Functions. Dlatego pole start_date w języku: default_args jest nieprzydatne.

Po uruchomieniu DAG wyświetla swoją konfigurację zgodnie ze zmienną print_gcs_info.

with airflow.DAG(
        'composer_sample_trigger_response_dag',
        default_args=default_args,
        # Not scheduled, trigger only
        schedule_interval=None) as dag:

    # Print the dag_run's configuration, which includes information about the
    # Cloud Storage object change.
    print_gcs_info = bash_operator.BashOperator(
        task_id='print_gcs_info', bash_command='echo {{ dag_run.conf }}')

6. Przetestuj funkcję

Otwórz środowisko Composer i w wierszu z nazwą środowiska kliknij link Airflow.

Otwórz composer_sample_trigger_response_dag, klikając jego nazwę. W tej chwili nie widać żadnych dowodów na uruchomienia DAG-ów, bo jeszcze ich nie uruchomiliśmy.Jeśli nie jest on widoczny ani klikalny, poczekaj minutę i odśwież stronę.

Otwórz oddzielną kartę i prześlij dowolny plik do utworzonego wcześniej zasobnika Cloud Storage określonego jako aktywatora funkcji w Cloud Functions. Możesz to zrobić za pomocą konsoli lub polecenia gsutil.

Wróć do karty z interfejsem Airflow i kliknij Widok wykresu

Kliknij zadanie print_gcs_info (zaznaczone na zielono).

Kliknij „Wyświetl dziennik”. w prawym górnym rogu menu

W logach zobaczysz informacje o pliku przesłanym przez Ciebie do zasobnika Cloud Storage.

Gratulacje! Właśnie udało Ci się aktywować DAG Airflow za pomocą Node.js i Google Cloud Functions.

7. Czyszczenie

Aby uniknąć obciążenia konta GCP opłatami za zasoby zużyte podczas krótkiego wprowadzenia:

  1. (Opcjonalnie) Aby zapisać swoje dane, pobierz je z zasobnika Cloud Storage środowiska Cloud Composer i zasobnika na dane utworzonego w ramach tego krótkiego wprowadzenia.
  2. Usuń zasobnik Cloud Storage dla utworzonego środowiska.
  3. Usuń środowisko Cloud Composer. Pamiętaj, że usunięcie środowiska nie powoduje usunięcia jego zasobnika na dane.
  4. (Opcjonalnie) W przypadku przetwarzania bezserwerowego pierwsze 2 miliony wywołań miesięcznie są bezpłatne, a gdy przeskalujesz swoją funkcję do zera, nie płacisz (więcej informacji znajdziesz w cenniku). Jeśli jednak chcesz usunąć swoją funkcję w Cloud Functions, kliknij „USUŃ” w prawym górnym rogu strony z omówieniem funkcji

4fe11e1b41b32ba2.png

Możesz też opcjonalnie usunąć projekt:

  1. W konsoli GCP otwórz stronę Projekty.
  2. Na liście projektów wybierz projekt do usunięcia, a następnie kliknij Usuń.
  3. W polu wpisz identyfikator projektu i kliknij Wyłącz, aby usunąć projekt.