Creación de un linaje de datos de extremo a extremo para consultas únicas y complejas mediante Amazon Athena, Amazon Redshift, Amazon Neptune y dbt

Las consultas únicas y complejas son dos escenarios comunes en el análisis de datos empresariales. Las consultas únicas son flexibles y adecuadas para el análisis instantáneo y la investigación exploratoria. Las consultas complejas, por otro lado, se refieren al procesamiento de datos a gran escala y al análisis en profundidad basado en almacenes de datos de nivel de petabyte en escenarios de datos masivos. Estas consultas complejas generalmente involucran fuentes de datos de múltiples sistemas empresariales, lo que requiere SQL anidado de varios niveles o asociaciones con numerosas tablas para tareas analíticas altamente sofisticadas.

Sin embargo, combinar el linaje de datos de estos dos tipos de consultas presenta varios desafíos:

1. Diversidad de fuentes de datos
2. Variación de la complejidad de las consultas
3. Granularidad inconsistente en el seguimiento del linaje
4. Diferentes requisitos en tiempo real
5. Dificultades en la integración entre sistemas

Además, mantener la precisión y la integridad de la información de linaje y, al mismo tiempo, garantizar el rendimiento y la escalabilidad del sistema son aspectos cruciales. Para abordar estos desafíos se requiere una arquitectura cuidadosamente diseñada y soluciones técnicas avanzadas.

Atenea amazónica ofrece análisis SQL flexible y sin servidor para consultas únicas, lo que permite la consulta directa de Servicio de almacenamiento simple de Amazon (Amazon S3) datos para un análisis instantáneo, rápido y rentable. Desplazamiento al rojo de Amazon, optimizado para consultas complejas, proporciona almacenamiento en columnas de alto rendimiento y arquitectura de procesamiento paralelo masivo (MPP), que admite procesamiento de datos a gran escala y capacidades SQL avanzadas. Amazonas Neptuno, como base de datos de gráficos, es ideal para el análisis de linaje de datos, ya que ofrece un recorrido de relaciones eficiente y algoritmos de gráficos complejos para manejar relaciones de linaje de datos complejas y de gran escala. La combinación de estos tres servicios proporciona una solución integral y poderosa para el análisis de linaje de datos de extremo a extremo.

En el contexto de una gobernanza integral de datos, Zona de datos de Amazon ofrece visualización del linaje de datos de toda la organización utilizando Amazon Web Services (AWS) servicios, mientras que dbt Proporciona linaje a nivel de proyecto a través del análisis de modelos y admite la integración entre proyectos, lagos de datos y almacenes.

En esta publicación, utilizamos dbt para el modelado de datos tanto en Amazon Athena como en Amazon Redshift. dbt en Athena admite consultas en tiempo real, mientras que dbt en Amazon Redshift maneja consultas complejas, unificando el lenguaje de desarrollo y reduciendo significativamente la curva de aprendizaje técnico. El uso de un solo lenguaje de modelado dbt no solo simplifica el proceso de desarrollo, sino que también genera automáticamente información de linaje de datos consistente. Este enfoque ofrece una adaptabilidad sólida, que se adapta fácilmente a los cambios en las estructuras de datos.

Al integrar la base de datos de gráficos de Amazon Neptune para almacenar y analizar relaciones de linaje complejas, combinadas con Funciones de paso de AWS y AWS Lambda Con estas funciones, logramos un proceso de generación de linaje de datos totalmente automatizado. Esta combinación promueve la consistencia y la integridad de los datos de linaje, al tiempo que mejora la eficiencia y la escalabilidad de todo el proceso. El resultado es una solución potente y flexible para el análisis de linaje de datos de extremo a extremo.

Descripción de la arquitectura

El contexto del experimento involucra a un cliente que ya utiliza Amazon Athena para consultas puntuales. Para adaptarse mejor al procesamiento masivo de datos y a escenarios de consultas complejas, su objetivo es adoptar un lenguaje de modelado de datos unificado en diferentes plataformas de datos. Esto llevó a la implementación de Athena en arquitecturas dbt y Amazon Redshift en arquitecturas dbt.

Pegamento AWS El rastreador rastrea la información del lago de datos de Amazon S3 y genera un catálogo de datos para respaldar el modelado de datos de dbt en Amazon Athena. Para escenarios de consultas complejas, AWS Glue realiza el procesamiento de extracción, transformación y carga (ETL), cargando los datos en el almacén de datos a escala de petabytes, Amazon Redshift. Aquí, el modelado de datos utiliza dbt en Amazon Redshift.

Los archivos originales de datos de linaje de ambas partes se cargan en un depósito S3, lo que proporciona soporte de datos para el análisis de linaje de datos de extremo a extremo.

La siguiente imagen es el diagrama de arquitectura de la solución.

Algunas consideraciones importantes:

Este experimento utiliza el siguiente diccionario de datos:

Los datos de linaje generados por dbt en Athena incluyen diagramas de linaje parciales, como se ejemplifica en las siguientes imágenes. La primera imagen muestra el linaje de name_basics En DBT sobre Atenea. La segunda imagen muestra el linaje de title_crew en dbt sobre Atenea.

Los datos de linaje generados por dbt en Amazon Redshift incluyen diagramas de linaje parciales, como se ilustra en la siguiente imagen.

Si nos basamos en el diccionario de datos y en las capturas de pantalla, resulta evidente que la información completa del linaje de datos está muy dispersa, repartida en 29 diagramas de linaje. Comprender la vista integral de extremo a extremo requiere mucho tiempo. En entornos del mundo real, la situación suele ser más compleja, ya que el linaje de datos completo puede estar distribuido en cientos de archivos. En consecuencia, la integración de un diagrama de linaje de datos completo de extremo a extremo se vuelve crucial y desafiante.

Este experimento proporcionará una introducción detallada al procesamiento y la fusión de archivos de linaje de datos almacenados en Amazon S3, como se ilustra en el siguiente diagrama.

Requisitos previos

Para ejecutar la solución, es necesario contar con los siguientes requisitos previos:

• La función Lambda para preprocesar archivos de linaje debe tener permisos para acceder a Amazon S3 y Amazon Redshift.
• La función Lambda para construir el Gráfico Acíclico Dirigido (DAG) debe tener permisos para acceder a Amazon S3 y Amazon Neptune.

Tutorial de la solución

Para realizar la solución siga los pasos de las siguientes secciones.

Preprocesamiento de datos de linaje sin procesar para la generación de DAG mediante funciones Lambda

Utilice Lambda para preprocesar los datos de linaje sin procesar generados por dbt, convirtiéndolos en archivos JSON de pares clave-valor que Neptune entiende fácilmente: athena_dbt_lineage_map.json y redshift_dbt_lineage_map.json.

1. Para crear una nueva función Lambda en la consola Lambda, ingrese un Nombre de la función, Haga click en el botón Runtime (Python en este ejemplo), configure el Arquitectura y Rol de ejecución, luego haga clic en "Crear función".

2. Abra la función Lambda creada y en el Configuration pestaña, en el panel de navegación, seleccione Variables de entorno y elija sus configuraciones. Usando Athena en el procesamiento de dbt como ejemplo, configure las variables de entorno de la siguiente manera (el proceso para Amazon Redshift en dbt es similar):
   • INPUT_BUCKET: data-lineage-analysis-24-09-22 (reemplazar con la ruta del depósito S3 que almacena el Athena original en los archivos de linaje dbt)
   • INPUT_KEY: athena_manifest.json (el archivo de linaje original de Athena en dbt)
   • OUTPUT_BUCKET: data-lineage-analysis-24-09-22 (reemplazar con la ruta del depósito S3 para almacenar la salida preprocesada de Athena en los archivos de linaje dbt)
   • OUTPUT_KEY: athena_dbt_lineage_map.json (el archivo de salida después de preprocesar el archivo de linaje Athena en dbt original)

3. En la pestaña Código En la pestaña, en el archivo lambda_function.py, ingrese el código de preprocesamiento para los datos de linaje sin procesar. Aquí hay una referencia de código que utiliza Athena en el procesamiento de dbt como ejemplo (el proceso para Amazon Redshift en dbt es similar). El código de preprocesamiento para Athena en el archivo de linaje original de dbt es el siguiente:

La athena_manifest.json, redshift_manifest.json, y otros archivos utilizados en este experimento se pueden obtener de Repositorio de GitHub de construcción de gráficos de linaje de datos.

import json
import boto3
import os

def lambda_handler(event, context):
    # Set up S3 client
    s3 = boto3.client('s3')

    # Get input and output paths from environment variables
    input_bucket = os.environ['INPUT_BUCKET']
    input_key = os.environ['INPUT_KEY']
    output_bucket = os.environ['OUTPUT_BUCKET']
    output_key = os.environ['OUTPUT_KEY']

    # Define helper function
    def dbt_nodename_format(node_name):
        return node_name.split(".")[-1]

    # Read input JSON file from S3
    response = s3.get_object(Bucket=input_bucket, Key=input_key)
    file_content = response['Body'].read().decode('utf-8')
    data = json.loads(file_content)
    lineage_map = data["child_map"]
    node_dict = {}
    dbt_lineage_map = {}

    # Process data
    for item in lineage_map:
        lineage_map[item] = [dbt_nodename_format(child) for child in lineage_map[item]]
        node_dict[item] = dbt_nodename_format(item)

    # Update key names
    lineage_map = {node_dict[old]: value for old, value in lineage_map.items()}
    dbt_lineage_map["lineage_map"] = lineage_map

    # Convert result to JSON string
    result_json = json.dumps(dbt_lineage_map)

    # Write JSON string to S3
    s3.put_object(Body=result_json, Bucket=output_bucket, Key=output_key)
    print(f"Data written to s3://{output_bucket}/{output_key}")

    return {
        'statusCode': 200,
        'body': json.dumps('Athena data lineage processing completed successfully')
    }

Fusionar datos de linaje preprocesados ​​y escribir en Neptune usando funciones Lambda

1. Antes de procesar datos con la función Lambda, cree una capa Lambda cargando el complemento Gremlin requerido. Para conocer los pasos detallados sobre la creación y configuración de Lambda capas, Ver el Documentación de capas de AWS Lambda.

Dado que conectar Lambda a Neptune para construir un DAG requiere el complemento Gremlin, es necesario cargarlo antes de usar Lambda. El paquete Gremlin se puede obtener en Repositorio de GitHub de construcción de gráficos de linaje de datos.

2. Crea una nueva función Lambda. Elige la función que quieres configurar. En la capa recién creada, en la parte inferior de la página, elige Agregar una capa.

Cree otra capa Lambda para la biblioteca de solicitudes, de forma similar a cómo creó la capa para el complemento Gremlin. Esta biblioteca se utilizará para la funcionalidad del cliente HTTP en la función Lambda.

3. Seleccione la función Lambda creada recientemente para configurarla. Conéctese a Neptune a través de Lambda para fusionar los dos conjuntos de datos y construir un DAG. Código pestaña, el código de referencia a ejecutar es el siguiente:

import json
import boto3
import os
import requests
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.credentials import get_credentials
from botocore.session import Session
from concurrent.futures import ThreadPoolExecutor, as_completed

def read_s3_file(s3_client, bucket, key):
    try:
        response = s3_client.get_object(Bucket=bucket, Key=key)
        data = json.loads(response['Body'].read().decode('utf-8'))
        return data.get("lineage_map", {})
    except Exception as e:
        print(f"Error reading S3 file {bucket}/{key}: {str(e)}")
        raise

def merge_data(athena_data, redshift_data):
    return {**athena_data, **redshift_data}

def sign_request(request):
    credentials = get_credentials(Session())
    auth = SigV4Auth(credentials, 'neptune-db', os.environ['AWS_REGION'])
    auth.add_auth(request)
    return dict(request.headers)

def send_request(url, headers, data):
    try:
        response = requests.post(url, headers=headers, data=data, timeout=30)
        response.raise_for_status()
        return response.text
    except requests.exceptions.RequestException as e:
        print(f"Request Error: {str(e)}")
        if hasattr(e.response, 'text'):
            print(f"Response content: {e.response.text}")
        raise

def write_to_neptune(data):
    endpoint = 'https://your neptune endpoint name:8182/gremlin'
    # replace with your neptune endpoint name

    # Clear Neptune database
    clear_query = "g.V().drop()"
    request = AWSRequest(method='POST', url=endpoint, data=json.dumps({'gremlin': clear_query}))
    signed_headers = sign_request(request)
    response = send_request(endpoint, signed_headers, json.dumps({'gremlin': clear_query}))
    print(f"Clear database response: {response}")

    # Verify if the database is empty
    verify_query = "g.V().count()"
    request = AWSRequest(method='POST', url=endpoint, data=json.dumps({'gremlin': verify_query}))
    signed_headers = sign_request(request)
    response = send_request(endpoint, signed_headers, json.dumps({'gremlin': verify_query}))
    print(f"Vertex count after clearing: {response}")
    
    def process_node(node, children):
        # Add node
        query = f"g.V().has('lineage_node', 'node_name', '{node}').fold().coalesce(unfold(), addV('lineage_node').property('node_name', '{node}'))"
        request = AWSRequest(method='POST', url=endpoint, data=json.dumps({'gremlin': query}))
        signed_headers = sign_request(request)
        response = send_request(endpoint, signed_headers, json.dumps({'gremlin': query}))
        print(f"Add node response for {node}: {response}")

        for child_node in children:
            # Add child node
            query = f"g.V().has('lineage_node', 'node_name', '{child_node}').fold().coalesce(unfold(), addV('lineage_node').property('node_name', '{child_node}'))"
            request = AWSRequest(method='POST', url=endpoint, data=json.dumps({'gremlin': query}))
            signed_headers = sign_request(request)
            response = send_request(endpoint, signed_headers, json.dumps({'gremlin': query}))
            print(f"Add child node response for {child_node}: {response}")

            # Add edge
            query = f"g.V().has('lineage_node', 'node_name', '{node}').as('a').V().has('lineage_node', 'node_name', '{child_node}').coalesce(inE('lineage_edge').where(outV().as('a')), addE('lineage_edge').from('a').property('edge_name', ' '))"
            request = AWSRequest(method='POST', url=endpoint, data=json.dumps({'gremlin': query}))
            signed_headers = sign_request(request)
            response = send_request(endpoint, signed_headers, json.dumps({'gremlin': query}))
            print(f"Add edge response for {node} -> {child_node}: {response}")

    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(process_node, node, children) for node, children in data.items()]
        for future in as_completed(futures):
            try:
                future.result()
            except Exception as e:
                print(f"Error in processing node: {str(e)}")

def lambda_handler(event, context):
    # Initialize S3 client
    s3_client = boto3.client('s3')

    # S3 bucket and file paths
    bucket_name = 'data-lineage-analysis' # Replace with your S3 bucket name
    athena_key = 'athena_dbt_lineage_map.json' # Replace with your athena lineage key value output json name
    redshift_key = 'redshift_dbt_lineage_map.json' # Replace with your redshift lineage key value output json name

    try:
        # Read Athena lineage data
        athena_data = read_s3_file(s3_client, bucket_name, athena_key)
        print(f"Athena data size: {len(athena_data)}")

        # Read Redshift lineage data
        redshift_data = read_s3_file(s3_client, bucket_name, redshift_key)
        print(f"Redshift data size: {len(redshift_data)}")

        # Merge data
        combined_data = merge_data(athena_data, redshift_data)
        print(f"Combined data size: {len(combined_data)}")

        # Write to Neptune (including clearing the database)
        write_to_neptune(combined_data)

        return {
            'statusCode': 200,
            'body': json.dumps('Data successfully written to Neptune')
        }
    except Exception as e:
        print(f"Error in lambda_handler: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps(f'Error: {str(e)}')
        }

Crear flujo de trabajo de Step Functions

1. En la consola Step Functions, elija Máquinas de estadoy luego elige Crear máquina de estado. Por Elija una plantilla página, seleccione Plantilla en blanco.

2. En Los Plantilla en blanco, escoger Código Para definir su máquina de estados, utilice el siguiente código de ejemplo:

{
  "Comment": "Daily Data Lineage Processing Workflow",
  "StartAt": "Parallel Processing",
  "States": {
    "Parallel Processing": {
      "Type": "Parallel",
      "Branches": [
        {
          "StartAt": "Process Athena Data",
          "States": {
            "Process Athena Data": {
              "Type": "Task",
              "Resource": "arn:aws:states:::lambda:invoke",
              "Parameters": {
                "FunctionName": "athena-data-lineange-process-Lambda", ##Replace with your Athena data lineage process Lambda function name
                "Payload": {
                  "input.$": "$"
                }
              },
              "End": true
            }
          }
        },
        {
          "StartAt": "Process Redshift Data",
          "States": {
            "Process Redshift Data": {
              "Type": "Task",
              "Resource": "arn:aws:states:::lambda:invoke",
              "Parameters": {
                "FunctionName": "redshift-data-lineange-process-Lambda", ##Replace with your Redshift data lineage process Lambda function name
                "Payload": {
                  "input.$": "$"
                }
              },
              "End": true
            }
          }
        }
      ],
      "Next": "Load Data to Neptune"
    },
    "Load Data to Neptune": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "FunctionName": "data-lineage-analysis-lambda" ##Replace with your Lambda function Name
      },
      "End": true
    }
  }
}

3. Después de completar la configuración, seleccione la opción Diseño Pestaña para ver el flujo de trabajo que se muestra en el siguiente diagrama.

Crear reglas de programación con Amazon EventBridge

Configure Amazon EventBridge para generar datos de linaje diariamente durante las horas de menor actividad comercial. Para ello:

1. Cree una nueva regla en la consola EventBridge con un nombre descriptivo.
2. Establezca el tipo de regla en “Programa” y configúrelo para que se ejecute una vez al día (utilizando una tasa fija o la expresión Cron “0 0 * * ? *”).
3. Seleccione la máquina de estados de AWS Step Functions como destino y especifique la máquina de estados que creó anteriormente.

Resultados de la consulta en Neptune

1. En la consola Neptune, seleccione Cuadernos. Abra un cuaderno existente o cree uno nuevo.

2. En el cuaderno, cree una nueva celda de código para realizar una consulta. El siguiente ejemplo de código muestra la declaración de consulta y sus resultados:

%%gremlin -d node_name -de edge_name
g.V().hasLabel('lineage_node').outE('lineage_edge').inV().hasLabel('lineage_node').path().by(elementMap())

Ahora puede ver la información del gráfico de linaje de datos de extremo a extremo tanto para dbt en Athena como para dbt en Amazon Redshift. La siguiente imagen muestra el gráfico de linaje de datos de DAG fusionado en Neptune.

Puede consultar el gráfico de linaje de datos generado para obtener datos relacionados con una tabla específica, como title_crew.

La declaración de consulta de muestra y sus resultados se muestran en el siguiente ejemplo de código:

%%gremlin -d node_name -de edge_name
g.V().has('lineage_node', 'node_name', 'title_crew')
  .repeat(
    union(
      __.inE('lineage_edge').outV(),
      __.outE('lineage_edge').inV()
    )
  )
  .until(
    __.has('node_name', within('names', 'genre_titles', 'titles'))
    .or()
    .loops().is(gt(10))
  )
  .path()
  .by(elementMap())

La siguiente imagen muestra los resultados filtrados según la tabla title_crew en Neptune.

Limpiar

Para limpiar sus recursos, complete los siguientes pasos:

1. Eliminar reglas de EventBridge

# Stop new events from triggering while removing dependencies
aws events disable-rule --name <rule-name>
# Break connections between rule and targets (like Lambda functions)
aws events remove-targets --rule <rule-name> --ids <target-id>
# Remove the rule completely from EventBridge
aws events delete-rule --name <rule-name>

2. Eliminar la máquina de estados de Step Functions

# Stop all running executions
aws stepfunctions stop-execution --execution-arn <execution-arn>
# Delete the state machine
aws stepfunctions delete-state-machine --state-machine-arn <state-machine-arn>

3. Eliminar funciones Lambda

# Delete Lambda function
aws lambda delete-function --function-name <function-name>
# Delete Lambda layers (if used)
aws lambda delete-layer-version --layer-name <layer-name> --version-number <version>

4. Limpiar la base de datos de Neptune

# Delete all snapshots
aws neptune delete-db-cluster-snapshot --db-cluster-snapshot-identifier <snapshot-id>
# Delete database instance
aws neptune delete-db-instance --db-instance-identifier <instance-id> --skip-final-snapshot
# Delete database cluster
aws neptune delete-db-cluster --db-cluster-identifier <cluster-id> --skip-final-snapshot

5. Siga las instrucciones en Eliminar un solo objeto Para limpiar los depósitos S3

Conclusión

En esta publicación, demostramos cómo dbt permite un modelado de datos unificado en Amazon Athena y Amazon Redshift, integrando el linaje de datos de consultas únicas y complejas. Al utilizar Amazon Neptune, esta solución proporciona un análisis de linaje integral de extremo a extremo. La arquitectura utiliza computación sin servidor de AWS y servicios administrados, incluidos Step Functions, Lambda y EventBridge, lo que proporciona un diseño altamente flexible y escalable.

Este enfoque reduce significativamente la curva de aprendizaje a través de un método de modelado de datos unificado y, al mismo tiempo, mejora la eficiencia del desarrollo. La visualización y el análisis de gráficos de linaje de datos de extremo a extremo no solo fortalecen las capacidades de gobernanza de datos, sino que también ofrecen información detallada para la toma de decisiones.

La arquitectura flexible y escalable de la solución optimiza eficazmente los costos operativos y mejora la capacidad de respuesta del negocio. Este enfoque integral equilibra la innovación técnica, la gobernanza de datos, la eficiencia operativa y la rentabilidad, lo que respalda el crecimiento empresarial a largo plazo con la adaptabilidad para satisfacer las cambiantes necesidades empresariales.

Con linaje de datos compatible con OpenLineage ahora Generalmente disponible en Amazon DataZonePlaneamos explorar posibilidades de integración para mejorar aún más la capacidad del sistema para manejar escenarios complejos de análisis de linaje de datos.

Si tiene alguna pregunta, no dude en dejar un comentario en la sección de comentarios.

Sobre los autores

nancy wu es arquitecto de soluciones en AWS, responsable de la consultoría y el diseño de arquitectura de computación en la nube para clientes empresariales multinacionales. Tiene muchos años de experiencia en big data, investigación y desarrollo de transformación digital empresarial, consultoría y gestión de proyectos en las industrias de telecomunicaciones, entretenimiento y finanzas.

Xu Feng es arquitecto sénior de soluciones industriales en AWS, responsable de diseñar, crear y promover soluciones industriales para los sectores de medios, entretenimiento y publicidad, como servicio al cliente inteligente e inteligencia empresarial. Con 20 años de experiencia en la industria del software, actualmente se centra en la investigación e implementación de IA generativa y soluciones de datos impulsadas por IA.

Xu Da es un arquitecto de soluciones para socios de Amazon Web Services (AWS) radicado en Shanghái, China. Tiene más de 25 años de experiencia en la industria de TI, desarrollo de software y arquitectura de soluciones. Le apasiona el aprendizaje colaborativo, el intercambio de conocimientos y la orientación de la comunidad en su recorrido hacia las tecnologías de la nube.

• Distribución de relaciones públicas y contenido potenciado por SEO. Consiga amplificado hoy.
• PlatoData.Network Vertical Generativo Ai. Empodérate. Accede Aquí.
• PlatoAiStream. Inteligencia Web3. Conocimiento amplificado. Accede Aquí.
• PlatoESG. Carbón, tecnología limpia, Energía, Ambiente, Solar, Gestión de residuos. Accede Aquí.
• PlatoSalud. Inteligencia en Biotecnología y Ensayos Clínicos. Accede Aquí.
• Fuente: https://aws.amazon.com/blogs/big-data/building-end-to-end-data-lineage-for-one-time-and-complex-queries-using-amazon-athena-amazon-redshift-amazon-neptune-and-dbt/