Google Cloud al modello Neo4j

Il modello Google Cloud a Neo4j consente di importare un set di dati in un database Neo4j tramite un job Dataflow, recuperando i dati da file CSV ospitati in bucket Cloud Storage. Consente anche di manipolare e trasformare i dati in varie fasi dell'importazione. Puoi utilizzare il modello sia per le importazioni iniziali che per quelle incrementali.

Requisiti della pipeline

• Un'istanza Neo4j in esecuzione
• Un bucket Cloud Storage
• Un set di dati da importare, sotto forma di file CSV
• Un file di specifica del job da utilizzare

Creare un file di specifiche del job

Il file delle specifiche del job è costituito da un oggetto JSON con le seguenti sezioni:

• config: flag globali che influiscono sulla modalità di esecuzione dell'importazione.
• sources: definizioni delle origini dati (relazionali).
• targets: definizioni della destinazione dei dati (grafico: nodi/relazioni).
• actions: azioni pre/post-caricamento.

Per ulteriori informazioni, consulta Creare un file di specifiche del job nella documentazione di Neo4j.

Parametri del modello

Parametri obbligatori

• jobSpecUri: il percorso del file di specifica del job, che contiene la descrizione JSON delle origini dati, delle destinazioni Neo4j e delle azioni.

Parametri facoltativi

• neo4jConnectionUri: il percorso del file JSON di connessione Neo4j.
• neo4jConnectionSecretId: l'ID secret per i metadati della connessione Neo4j. Puoi utilizzare questo valore in alternativa a neo4jConnectionUri.
• optionsJson: un oggetto JSON chiamato anche token di runtime. Ad esempio, {token1:value1,token2:value2}. Spec can refer to $token1 and $token2.. Il valore predefinito è vuoto.
• readQuery: override della query SQL. Il valore predefinito è vuoto.
• inputFilePattern: l'override del percorso del file di testo. Ad esempio, gs://your-bucket/path/*.json. Il valore predefinito è vuoto.
• disabledAlgorithms: algoritmi separati da virgole da disattivare. Se questo valore è impostato su none, nessun algoritmo viene disattivato. Utilizza questo parametro con cautela, perché gli algoritmi disabilitati per impostazione predefinita potrebbero presentare vulnerabilità o problemi di prestazioni. Ad esempio: SSLv3, RC4.
• extraFilesToStage: percorsi Cloud Storage o secret Secret Manager separati da virgole per i file da preparare nel worker. Questi file vengono salvati nella directory /extra_files di ogni worker. Ad esempio, gs://<BUCKET_NAME>/file.txt,projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<VERSION_ID>.

Esegui il modello

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. La regione predefinita è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

4. Nel menu a discesa Modello Dataflow, seleziona il modello Google Cloud a Neo4j.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota:per utilizzare Google Cloud CLI per eseguire modelli flex, devi avere la versione Google Cloud CLI 284.0.0 o successive.

Nella shell o nel terminale, esegui il modello:

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates/VERSION/flex/Google_Cloud_to_Neo4j \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --parameters \
       jobSpecUri=JOB_SPEC_URI,\
       neo4jConnectionUri=NEO4J_CONNECTION_URI,\

Sostituisci quanto segue:

• JOB_NAME: un nome univoco del job a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare l'ultima versione del modello, disponibile nella cartella principale senza data nel bucket: gs://dataflow-templates/latest/
  • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella principale con data nel bucket: gs://dataflow-templates/
  Attenzione:l'ultima versione dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale con data più recente per evitare che queste modifiche che causano interruzioni influiscano sui workflow di produzione.
• REGION_NAME: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• JOB_SPEC_URI: il percorso del file di specifica del job
• NEO4J_CONNECTION_URI: il percorso dei metadati della connessione Neo4j

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta HTTP POST. Per saperne di più sull'API e sui relativi ambiti di autorizzazione, consulta projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launchParameter": {
     "jobName": "JOB_NAME",
     "parameters": {
       "jobSpecUri": "JOB_SPEC_URI",
       "neo4jConnectionUri": "NEO4J_CONNECTION_URI",
     },
     "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/Google_Cloud_to_Neo4j",
     "environment": { "maxWorkers": "10" }
  }
}

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto Google Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome univoco del job a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare l'ultima versione del modello, disponibile nella cartella principale senza data nel bucket: gs://dataflow-templates/latest/
  • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella principale con data nel bucket: gs://dataflow-templates/
  Attenzione:l'ultima versione dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale con data più recente per evitare che queste modifiche che causano interruzioni influiscano sui workflow di produzione.
• LOCATION: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• JOB_SPEC_URI: il percorso del file di specifica del job
• NEO4J_CONNECTION_URI: il percorso dei metadati della connessione Neo4j

Codice sorgente del modello

Java

/*
 * Copyright (C) 2021 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.neo4j.templates;

import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.Template.AdditionalDocumentationBlock;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.neo4j.model.InputValidator;
import com.google.cloud.teleport.v2.neo4j.model.Json;
import com.google.cloud.teleport.v2.neo4j.model.Json.ParsingResult;
import com.google.cloud.teleport.v2.neo4j.model.connection.ConnectionParams;
import com.google.cloud.teleport.v2.neo4j.model.helpers.JobSpecMapper;
import com.google.cloud.teleport.v2.neo4j.model.helpers.OverlayTokenParser;
import com.google.cloud.teleport.v2.neo4j.options.Neo4jFlexTemplateOptions;
import com.google.cloud.teleport.v2.neo4j.utils.FileSystemUtils;
import com.google.cloud.teleport.v2.utils.SecretManagerUtils;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.commons.lang3.StringUtils;
import org.neo4j.importer.v1.pipeline.ImportPipeline;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Dataflow template which reads Google Cloud data (Text, BigQuery) and writes it to Neo4j.
 *
 * <p>In case of BigQuery, the source data can be either a table or a SQL query.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-neo4j/README_Google_Cloud_to_Neo4j.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Google_Cloud_to_Neo4j",
    category = TemplateCategory.BATCH,
    displayName = "Google Cloud to Neo4j",
    description =
        "The Google Cloud to Neo4j template lets you import a dataset into a Neo4j database through a Dataflow job, "
            + "sourcing data from CSV files hosted in Google Cloud Storage buckets. It also lets you to manipulate and transform the data "
            + "at various steps of the import. You can use the template for both first-time imports and incremental imports.",
    optionsClass = Neo4jFlexTemplateOptions.class,
    flexContainerName = "googlecloud-to-neo4j",
    contactInformation = "https://support.neo4j.com/",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/google-cloud-to-neo4j",
    requirements = {
      "A running Neo4j instance",
      "A Google Cloud Storage bucket",
      "A dataset to import, in the form of CSV files",
      "A job specification file to use"
    },
    additionalDocumentation = {
      @AdditionalDocumentationBlock(
          name = "Create a job specification file",
          content = {
            "The job specification file consists of a JSON object with the following sections:\n"
                + "- `config` - global flags affecting how the import is performed.\n"
                + "- `sources` - data source definitions (relational).\n"
                + "- `targets` - data target definitions (graph: nodes/relationships/custom queries).\n"
                + "- `actions` - pre/post-load actions.\n"
                + "For more information, see <a href=\"https://neo4j.com/docs/dataflow-google-cloud/job-specification/\" class=\"external\">Create a job specification file</a> in the Neo4j documentation."
          })
    },
    preview = true)
public class GoogleCloudToNeo4j {

  private static final Logger LOG = LoggerFactory.getLogger(GoogleCloudToNeo4j.class);

  /**
   * Runs a pipeline which reads data from various sources and writes it to Neo4j.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    Neo4jFlexTemplateOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(Neo4jFlexTemplateOptions.class);
    ValidationErrors.processValidations(
        "Errors found validating pipeline options: ",
        InputValidator.validateNeo4jPipelineOptions(options));

    if (StringUtils.isBlank(options.getDisabledAlgorithms())) {
      options.setDisabledAlgorithms(
          "SSLv3, RC4, DES, MD5withRSA, DH keySize < 1024, EC keySize < 224, 3DES_EDE_CBC, anon,"
              + " NULL");
    }
    LOG.info("Job: {}", options.getJobSpecUri());
    FileSystems.setDefaultPipelineOptions(options);
    var overlayTokens = OverlayTokenParser.parse(options.getOptionsJson());
    var templateVersion = readTemplateVersion(options);
    var neo4jConnectionConfig = readConnectionConfiguration(options);
    var importSpecification = JobSpecMapper.parse(options.getJobSpecUri(), overlayTokens);
    var importPipeline =
        new Neo4jImportPipelineRunner(
            options,
            templateVersion,
            neo4jConnectionConfig,
            overlayTokens,
            ImportPipeline.of(importSpecification));
    importPipeline.run();
  }

  private static String readTemplateVersion(Neo4jFlexTemplateOptions options) {
    var labels = options.as(DataflowPipelineOptions.class).getLabels();
    String defaultVersion = "UNKNOWN";
    if (labels == null) {
      return defaultVersion;
    }
    return labels.getOrDefault("goog-dataflow-provided-template-version", defaultVersion);
  }

  private static ConnectionParams readConnectionConfiguration(
      Neo4jFlexTemplateOptions pipelineOptions) {
    String neo4jConnectionJson = readConnectionSettings(pipelineOptions);
    ParsingResult parsingResult = InputValidator.validateNeo4jConnection(neo4jConnectionJson);
    if (!parsingResult.isSuccessful()) {
      ValidationErrors.processValidations(
          "Errors found validating Neo4j connection: ",
          parsingResult.formatErrors("Could not validate connection JSON"));
    }
    return Json.map(parsingResult, ConnectionParams.class);
  }

  private static String readConnectionSettings(Neo4jFlexTemplateOptions options) {
    String secretId = options.getNeo4jConnectionSecretId();
    if (StringUtils.isNotEmpty(secretId)) {
      return SecretManagerUtils.getSecret(secretId);
    }
    String uri = options.getNeo4jConnectionUri();
    try {
      return FileSystemUtils.getPathContents(uri);
    } catch (Exception e) {
      throw new RuntimeException(
          String.format("Unable to read Neo4j configuration at URI %s: ", uri), e);
    }
  }
}

Passaggi successivi

• Scopri di più sui modelli Dataflow.
• Consulta l'elenco dei modelli forniti da Google.