What’s required is a robust approach to constructing end-to-end information lineage for one-time and complex queries leveraging the capabilities of Amazon Athena, Amazon Redshift, Amazon Neptune, and dbt.

One-off and complex inquiries are two common scenarios in enterprise information analytics. One-time queries are remarkably flexible, making them well-suited for swift evaluation and preliminary exploration. Complex queries are efficiently resolved through large-scale information processing and in-depth evaluations leveraging massive data repositories in extensive information systems. Complex queries frequently involve data drawn from multiple enterprise applications, necessitating intricate multi-level nested SQL constructs or complex table associations to support highly nuanced analytical tasks?

Notwithstanding the complexity that arises from integrating the informational heritage of these two query formats, numerous obstacles present themselves.

1. Range of information sources
2. Various question complexity
3. Inconsistent granularity in lineage monitoring
4. Totally different real-time necessities
5. Difficulties in cross-system integration

To ensure the reliability and thoroughness of lineage information while also maintaining a high-performance computing environment is crucial. Overcoming these hurdles necessitates a meticulously crafted framework and cutting-edge technological solutions.

Provides serverless and versatile SQL analytics for one-off queries, allowing for fast and affordable direct querying of Amazon S3 data to facilitate instant analysis. Optimized for complex queries, the database leverages high-performance columnar storage and a massively parallel processing (MPP) architecture to efficiently support large-scale data processing and advanced SQL functionality. As a graph database, it’s particularly well-suited for information lineage evaluation, enabling environmentally friendly relationship traversal and complex graph algorithms to navigate large-scale, intricate information lineage relationships seamlessly. The combination of these three providers provides a comprehensive solution for thorough information lineage assessment from start to finish.

In the realm of comprehensive information governance, a solution provides organization-wide information lineage visualization leveraging AWS providers, whereas another tool offers project-level lineage through model evaluation and facilitates cross-project integration across data lakes and warehouses.

We leverage dbt to perform data modeling on both Amazon Athena and Amazon Redshift. Dbt on Athena empowers real-time query capabilities, while dbt on Amazon Redshift streamlines complex query handling, harmonizing the event language and significantly reducing the technical learning barrier. By leveraging a unified DBT modeling language, the data transformation process is streamlined, and automatically generated metadata tracks information lineage with consistency. This approach provides robust flexibility, seamlessly absorbing changes to data structures.

By leveraging Amazon Neptune’s advanced graph database capabilities, retailers can seamlessly integrate and analyze complex lineage relationships, combining these insights with existing features to deliver a fully automated information lineage process. This tool fosters cohesion and thoroughness in lineage data while amplifying efficiency and capacity for the entire process. The results provide a robust and versatile framework for comprehensive end-to-end information lineage evaluation.

Structure overview

This experiment’s context assumes a buyer who is already familiar with using Amazon Athena for occasional query executions. To effectively handle massive data processing and complex query scenarios, they plan to develop a standardized information modeling language across various information platforms. The successful outcome was the integration of each Athena on dbt architecture alongside Amazon Redshift on dbt infrastructure.

The crawler efficiently extracts valuable insights from the Amazon S3 repository, generating a comprehensive Knowledge Catalog that empowers data analysts to effectively model and analyze information stored in Amazon Athena. In complex data manipulation scenarios, AWS Glue efficiently executes extract, transform, and load (ETL) processes, seamlessly integrating information into a massive storage repository, Amazon Redshift. Information modeling uses dbt on Amazon Redshift to create a unified, analytics-ready data layer.

Unique lineage information from individual elements is uploaded to an Amazon S3 bucket, facilitating comprehensive information lineage analysis and enabling informed decision-making throughout the data lifecycle.

What kind of data do you want to visualize with this structure diagram?

Some vital concerns:

The purpose of this investigation relies on a pre-existing database.

The lineage data produced by dbt on Athena yields incomplete lineage diagrams, as illustrated in the accompanying images. The primary depiction discloses the genealogy of name_basics in dbt on Athena. The second photograph discloses the heritage of title_crew in dbt on Athena.

Dbt-generated lineage information for Amazon Redshift produces incomplete lineage diagrams, as depicted in the accompanying visual representation.

According to visual evidence from the information dictionary and screenshots, a sprawling complexity characterizes the entirety of the lineage data, with its fragmented nature spanning across 29 distinct diagrams. Acquiring a comprehensive understanding of the entire process necessitates a considerable investment of time. In everyday settings, complexity typically prevails, with data dissemination often spread across numerous sources. As a result, creating a comprehensive, end-to-end information lineage diagram becomes both crucial and challenging.

This experiment delves into processing and merging information lineage data stored in Amazon S3, as visualized in the accompanying diagram.

Conditions

To successfully execute the plan, it is essential that certain prerequisites are met.

• The Lambda function performing preprocessing on lineage information requires permission to access both Amazon S3 and Amazon Redshift.
• The Lambda function responsible for setting up a Directed Acyclic Graph (DAG) will require permissions to access Amazon S3 and Amazon Neptune resources.

Answer walkthrough

Complying with the steps outlined in the subsequent sections will enable you to successfully execute the task at hand.

Streamline processing of uncooked lineage data for efficient integration with DAG technology by leveraging the versatility of Lambda functions.

The following transformations are performed on the lineage information using Python’s `json` and `lambda` functions:

data = json.loads(lineage)
lineage_data = lambda x: {**{“root”:x[“root”]}, **{f”node_{i+1}”: {“name”:node[“name”], “parents”:node.get(“parents”,[]), “children”:node.get(“children”,[])} for i,node in enumerate(x.get(“nodes”,[]))}}(data) athena_dbt_lineage_map.json and redshift_dbt_lineage_map.json.

1. To create a brand-new Lambda function in the Lambda console, type a comma, select Python as the runtime, configure the handler and role, and then click the “Create function” button.

2. Select the Lambda function from the navigation pane and then configure your settings by clicking on the desired options. Configure Athena variables using dbt as follows to ensure seamless processing:
   • INPUT_BUCKET: data-lineage-analysis-24-09-22 The s3://dbt-athena-lineage-data/unique-athena-on-dbtl lineage information is stored.
   • INPUT_KEY: athena_manifest.json The definitive Athena on dbt lineage file.
   • OUTPUT_BUCKET: data-lineage-analysis-24-09-22 s3://dbt-lineage-data/preprocessed-athena-output/
   • OUTPUT_KEY: athena_dbt_lineage_map.json The processed Athena query output after processing the unique dbt lineage file for Athena.

3. In the Lambda function, insert the data processing logic for uncooked Lineage details on the specified Python file’s tab. Right here’s an example of a code reference using Athena on dbt processing, with an analogous method for Amazon Redshift on dbt. The pre-processing code for Athena on dbt’s unique lineage file is thus:

The athena_manifest.json, redshift_manifest.jsonDifferent information used on this experiment could potentially be obtained from various other sources.

import json
import boto3
import os

def lambda_handler(event, context):
    shopper = boto3.client('s3')

    input_bucket = os.environ['INPUT_BUCKET']
    input_key = os.environ['INPUT_KEY']
    output_bucket = os.environ['OUTPUT_BUCKET']
    output_key = os.environ['OUTPUT_KEY']

    def format_dbt_node_name(node_name):
        return node_name.split('.')[-1]

    response = shopper.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 = {}

    for item in lineage_map:
        lineage_map[item] = [format_dbt_node_name(child) for child in lineage_map[item]]
        node_dict[item] = format_dbt_node_name(item)

    lineage_map = {node_dict[old]: new for old, new in lineage_map.items()}
    dbt_lineage_map['lineage_map'] = lineage_map

    result_json = json.dumps(dbt_lineage_map)
    shopper.put_object(Body=result_json.encode('utf-8'), 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 accomplished efficiently') }

Can we extract lineage metadata and directly inject into Neptune using their Lambda functions?

1. Before utilizing the Lambda function to process data, create a Lambda layer that includes the necessary Gremlin plugin for importation. To create and configure Amazon Lambda, refer to the documentation.

Connecting Lambda to Neptune for setting up a Directed Acyclic Graph (DAG) necessitates uploading the Gremlin plugin beforehand, as this step is required prior to utilizing Lambda. The GRANITE package can be obtained from the CRAN.

2. What’s the best way to create a new lambda function?

   1. Log into the AWS Management Console.
   2. Navigate to the Lambda dashboard and click on “Create function”.
   3. Select “Author from scratch” as the template type.
   4. Give your function a name, select a runtime (e.g., Node.js or Python), and choose the execution role for your function.
   5. Configure the environment variables, if needed.
   6. Define the handler and the entry point of your function.
   7. Set the memory size and timeout values for your function.
   8. Click “Create function” to deploy your new lambda.

   SKIP Select the perform to configure. On the newly created layer at the back of the webpage, click .

Can we create another lambda layer for the requests library? This library can be utilised for HTTP shopper performance within the AWS Lambda function.

3. Recently established Lambda functions need to be set up properly. Merge the two datasets leveraging Neptune’s Lambda functionality, ultimately crafting a Directed Acyclic Graph (DAG). On the tab, the reference code to be executed is specified as follows:

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):
    strive:
        response = s3_client.get_object(Bucket=bucket, Key=key)
        information = json.masses(response['Body'].learn().decode('utf-8'))
        return information.get("lineage_map", {})
    besides Exception as e:
        print(f"Error studying S3 file {bucket}/{key}: {str(e)}")
        elevate

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, information):
    strive:
        response = requests.publish(url, headers=headers, information=information, timeout=30)
        response.raise_for_status()
        return response.textual content
    besides requests.exceptions.RequestException as e:
        print(f"Request Error: {str(e)}")
        if hasattr(e.response, 'textual content'):
            print(f"Response content material: {e.response.textual content}")
        elevate

def write_to_neptune(information):
    endpoint="https://your neptune endpoint identify:8182/gremlin"
    # change along with your neptune endpoint identify

    # Clear Neptune database
    clear_query = "g.V().drop()"
    request = AWSRequest(technique='POST', url=endpoint, information=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}")

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

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

            # Add edge
            question = f"g.V().has('lineage_node', 'node_name', '{node}').as('a').V().has('lineage_node', 'node_name', '{child_node}').coalesce(inE('lineage_edge').the place(outV().as('a')), addE('lineage_edge').from('a').property('edge_name', ' '))"
            request = AWSRequest(technique='POST', url=endpoint, information=json.dumps({'gremlin': question}))
            signed_headers = sign_request(request)
            response = send_request(endpoint, signed_headers, json.dumps({'gremlin': question}))
            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):
            strive:
                future.consequence()
            besides Exception as e:
                print(f"Error in processing node: {str(e)}")

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

    # S3 bucket and file paths
    bucket_name="data-lineage-analysis" # Substitute along with your S3 bucket identify
    athena_key = 'athena_dbt_lineage_map.json' # Substitute along with your athena lineage key worth output json identify
    redshift_key = 'redshift_dbt_lineage_map.json' # Substitute along with your redshift lineage key worth output json identify

    strive:
        # Learn Athena lineage information
        athena_data = read_s3_file(s3_client, bucket_name, athena_key)
        print(f"Athena information dimension: {len(athena_data)}")

        # Learn Redshift lineage information
        redshift_data = read_s3_file(s3_client, bucket_name, redshift_key)
        print(f"Redshift information dimension: {len(redshift_data)}")

        # Merge information
        combined_data = merge_data(athena_data, redshift_data)
        print(f"Mixed information dimension: {len(combined_data)}")

        # Write to Neptune (together with clearing the database)
        write_to_neptune(combined_data)

        return {
            'statusCode': 200,
            'physique': json.dumps('Knowledge efficiently written to Neptune')
        }
    besides Exception as e:
        print(f"Error in lambda_handler: {str(e)}")
        return {
            'statusCode': 500,
            'physique': json.dumps(f'Error: {str(e)}')
        }

Create Step Capabilities workflow

1. In the Step Capabilities console, choose the option followed by clicking. In the website’s navigation menu, select the desired option.

2. Let’s design a state machine for an ATM system.

   **State Diagram:**

   “`
   +—————+
   | Idle |
   +—————+
   |
   | Insert Card
   v
   +—————+
   | Login |
   +—————+
   |
   | Enter PIN
   v
   +—————+
   | Authenticated|
   +—————+
   |
   | Select Option
   v
   +—————+
   | Withdrawal |
   | Deposit |
   | Balance |
   | Exit |
   +—————+
   “`

   SKIP Use the next instance code:

{
  "Remark": "Every Day Knowledge Lineage Processing Workflow",
  "StartAt": "Parallel Data Processing",
  "States": {
    "Parallel Data Processing": {
      "Type": "Map",
      "Items": "$",
      "ResultPath": "$.data",
      "Next": "Knowledge Loading",
      "States": {
        "Process Athena Data": {
          "Type": "Task",
          "Resource": "arn:aws:lambda:REGION:ACCOUNT_ID:function:AthenaDataLineageProcess-Lambda",
          "Parameters": {
            "input.$": "$.data"
          },
          "End": true
        },
        "Process Redshift Data": {
          "Type": "Task",
          "Resource": "arn:aws:lambda:REGION:ACCOUNT_ID:function:RedshiftDataLineageProcess-Lambda",
          "Parameters": {
            "input.$": "$.data"
          },
          "End": true
        }
      }
    },
    "Knowledge Loading": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:REGION:ACCOUNT_ID:function:DataLineageAnalysis-Lambda",
      "Parameters": {
        "input.$": "$.data"
      },
      "End": true
    }
  }
}

3. Upon completing the configuration, navigate to the relevant tab to visualize the workflows illustrated in a clear and concise diagram.

What are the scheduling guidelines for Amazon EventBridge that ensure reliable event processing and minimize downtime?

To schedule events effectively in Amazon EventBridge, consider the following best practices:

1.? Establish a consistent scheduling frequency: Determine the optimal interval for processing events, taking into account factors like system load, data volume, and business requirements.

2.? Utilize Schedule Expressions: Leverage Schedule Expressions to define complex schedules that cater to your specific event processing needs. This feature enables you to create custom schedules using a variety of built-in functions and time zones.

3.? Set up Event Filters: Implement Event Filters to prioritize events based on criteria such as event type, source, or priority. This helps ensure that critical events are processed promptly while less urgent events can be delayed if necessary.

4.? Configure Dead Letter Queues (DLQs): Establish DLQs to handle failed events and prevent message loss. This feature enables you to store failed events for analysis and debugging purposes.

5.? Monitor and Adjust: Continuously monitor event processing performance and adjust scheduling frequencies, filter settings, or other parameters as needed to maintain optimal system efficiency.

By following these guidelines, you can create a robust and scalable event-driven architecture that meets your organization’s specific needs.

Configure Amazon EventBridge to capture and store lineage data daily during a designated maintenance window, ensuring accurate tracking of business processes and decision-making insights. To do that:

1. You create a brand new rule within the AWS EventBridge console by navigating to the Rules tab, clicking on Create event rule, and then specifying a descriptive name for your rule.
2. ? At 12:00 AM daily, execute this task when the system is running, provided that there are no errors from the previous attempt. *”).
3. We’re selecting the AWS Step Function Capabilities state machine since that aligns with our objective, utilizing the workflow we previously developed.

Question leads to Neptune

1. On the Neptune console, select. Begin a fresh journey with an open notebook or create a blank slate in a brand-new one.

2. In the newly created code cell, write down your query in plain language, followed by a colon and a blank line, like so:

   What is the average airspeed velocity of an unladen swallow: The questions being asserted are unclear and require more context to understand. Can you provide additional information about the purpose of these assertions?

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

Now you can visualize the end-to-end information lineage graph for each DBT model on both Athena and Amazon Redshift, providing enhanced transparency and insights into your data pipeline. The subsequent visualisation showcases a consolidated Directed Acyclic Graph (DAG) detailing the information lineage within Neptune’s framework.

You will be able to query the generated information lineage graph for information linked to a specific desk, similar to title_crew.

The logical relationship between a question, an assertion, and their outcomes is demonstrated in this specific code example.

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

The subsequent image displays filtered results primarily driven by the title_crew desk situated in Neptune.

Clear up

To maximize the value of your assets, follow these next steps:

1. Delete EventBridge guidelines

# Cease new occasions from triggering whereas eradicating dependencies
aws occasions disable-rule --name <rule-name>
# Break connections between rule and targets (like Lambda features)
aws occasions remove-targets --rule <rule-name> --ids <target-id>
# Take away the rule fully from EventBridge
aws occasions delete-rule --name <rule-name>

2. Delete Step Capabilities state machine

# Cease all working 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. Delete Lambda features

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

4. Clear up the Neptune database

# Delete all snapshots
aws neptune delete-db-cluster-snapshot --db-cluster-snapshot-identifier <snapshot-id>
# Delete database occasion
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. To properly wash up the S3 buckets:

   1. Open AWS Management Console and navigate to the S3 dashboard.
   2. Select the bucket that needs cleaning up from the list of available buckets.
   3. Click on ‘Properties’ tab in the navigation panel.
   4. Look for the ‘Lifecycle rule’ section. If no rules are set, click on ‘Edit’ to create a new rule or modify an existing one.
   5. Set the lifecycle rule to delete objects older than 30 days and specify the date when you want this action to start.
   6. Click ‘Save changes’ to confirm your actions.

   SKIP

Conclusion

On this publication, we showcased how dbt enables unified data modeling across Amazon Athena and Amazon Redshift, harmonizing information lineage from both simple and complex queries. With Amazon Neptune, this solution provides comprehensive, end-to-end lineage assessment capabilities. The architecture leverages the benefits of AWS serverless computing and managed services, combining Step Functions, Lambda, and EventBridge to create a highly adaptable and scalable framework.

This approach dramatically reduces the educational hurdle through a single, cohesive framework for information management, thereby amplifying effectiveness and promoting sustainable growth. The visual representation of end-to-end information lineage, complemented by evaluation tools, bolsters organizational governance while providing actionable insights to inform strategic decisions.

The answer’s adaptable framework effectively streamlines process costs, boosting organisational agility and reaction times. This comprehensive strategy harmonizes technical innovation, information governance, operational efficiency, and cost-effectiveness, thereby enabling long-term business growth while accommodating evolving enterprise requirements.

As OpenLineage compatibility is now in place, our objective is to uncover opportunities for integrations that will further enhance the system’s capabilities, ultimately enabling more effective management of complex information lineage assessment scenarios.

What would you like to ask?

Concerning the authors

Serving as an Options Architect at AWS, I am responsible for crafting cloud computing infrastructure solutions tailored to the unique needs of large-scale enterprise clients. Boasting extensive experience across multiple sectors, including telecommunications, entertainment, and finance, I possess several years of expertise in analyzing and driving large-scale digital transformations, strategic growth initiatives, and management consulting projects.

is a Senior Business Answer Architect at AWS, accountable for designing, constructing, and selling trade options for the Media & Leisure and Promoting sectors, akin to clever customer support and enterprise intelligence. With two decades of expertise in the software development industry, currently focused on researching and deploying generative AI and AI-infused data solutions.

As an AWS Associate Options Architect based in Shanghai, China. With over 25 years of extensive experience in the IT sector, software development, and architecture. He’s intensely committed to fostering a collaborative environment where individuals can learn from each other, share knowledge, and navigate the complexities of cloud technologies together.