Column-Level Lineage Diff on BitBucket

Set up Column-Level Lineage Diff on BitBucket. First complete the shared Lineage Diff prerequisites, then configure your BitBucket pipeline.

Generate API keys

Generate a Paradime API key with the Bolt schedules admin capability. Use an Account API key (a prdm_cmp_... Bearer token), and note your workspace token and API endpoint; you will add them as CI/CD variables in a later step.

Generate API keys

Create a BitBucket Pipeline

Now you will need to create a new bitbucket-pipelines.yml file in your dbt™️ repository. Copy the code block below. This pipeline has two steps:

Run Paradime Turbo CI — Runs the Turbo CI bolt schedule and saves the BOLT_RUN_ID as a pipeline artifact.
Generate Lineage Diff Report — Reads the BOLT_RUN_ID from the artifact and triggers the lineage diff report.

image: python:3.11

definitions:
  steps:
    - step: &paradime-turbo-ci
        name: Run Paradime Turbo CI
        size: 2x
        max-time: 60
        script:
          - apt-get update && apt-get install -y git
          - pip install paradime-io
          - export PARADIME_SCHEDULE_SLUG="flowing-pachy-wyy4fs"
          - echo "Merge Request Number: $BITBUCKET_PR_ID"
          - echo "Commit SHA: $BITBUCKET_COMMIT"
          - |
            cat > run_turbo_ci.py << 'EOL'
            import os
            from paradime import Paradime

            paradime = Paradime(
              workspace_uid=os.environ['PARADIME_WORKSPACE_UID'],
              api_secret=os.environ['PARADIME_API_SECRET'],
              api_endpoint=os.environ['PARADIME_API_ENDPOINT']
            )

            run_id = paradime.bolt.trigger_run(
              slug=os.environ['PARADIME_SCHEDULE_SLUG'],
              branch=os.environ['BITBUCKET_COMMIT'],
              pr_number=int(os.environ['BITBUCKET_PR_ID']),
            )

            print(f"Bolt Run ID: {run_id}")

            with open("bolt.env", "w") as f:
              f.write(f"BOLT_RUN_ID={run_id}\n")
            EOL
          - python3 run_turbo_ci.py
        artifacts:
          - bolt.env

    - step: &paradime-lineage-diff-report
        name: Generate Lineage Diff Report
        size: 2x
        max-time: 30
        script:
          - apt-get update && apt-get install -y curl jq git
          - pip install paradime-io
          - source bolt.env
          - echo "Bolt Run ID: $BOLT_RUN_ID"
          - git fetch origin $BITBUCKET_PR_DESTINATION_BRANCH
          - export HEAD_COMMIT_SHA=$(git rev-parse $BITBUCKET_PR_DESTINATION_COMMIT)
          - export BASE_COMMIT_SHA=$BITBUCKET_COMMIT
          - export PULL_REQUEST_NUMBER=$BITBUCKET_PR_ID
          - export USER_EMAIL=$(git log -1 --pretty=format:"%ae" $BITBUCKET_COMMIT)
          - export CHANGED_FILES=$(git diff --name-only $BASE_COMMIT_SHA $HEAD_COMMIT_SHA)
          - echo "Pull Request Number $PULL_REQUEST_NUMBER"
          - echo "Pull Request Author Email $USER_EMAIL"
          - echo "Files changed in this PR $CHANGED_FILES"
          - export REPORT_FILE="report_$(openssl rand -hex 12).json"
          - echo "Report file $REPORT_FILE"
          - |
            cat > generate_report.py << 'EOL'
            import os
            import json
            from pathlib import Path
            from paradime import Paradime

            def format_json_to_markdown(report_json):
                """
                Convert the JSON report into formatted markdown for Bitbucket comments.
                """
                formatted_output = []

                # 1. Title and Warning
                formatted_output.append("# Paradime Column-Level Lineage Analysis\n")
                formatted_output.append("⚠️ The changes included in this pull request may have downstream impacts:\n")

                # 2. Table headers and data
                formatted_output.append("| File changed | dbt nodes | dbt projects | Tableau | Looker | ThoughtSpot |")
                formatted_output.append("| :- | -: | -: | -: | -: | -: |")

                # Process each file's impact summary
                for file_path, file_data in report_json["changed_files"].items():
                    summary = file_data["impact_summary"]
                    row = [
                        file_path,
                        f"**{summary['dbt_nodes_count']}**",
                        f"**{summary['dbt_projects_count']}**",
                        f"**{summary['tableau_nodes_count']}**",
                        f"**{summary['looker_nodes_count']}**",
                        f"**{summary['thoughtspot_nodes_count']}**"
                    ]
                    formatted_output.append(f"| {' | '.join(row)} |")

                # 3. Impact Details Section
                formatted_output.append("\n## 📁 Impact by Changed Files\n")
                formatted_output.append("Expand to view the potential downstream impact created by each changed file:\n")

                # Process each file's detailed impacts
                for file_path, file_data in report_json["changed_files"].items():
                    # File header
                    formatted_output.append(f"\n### {file_path}\n")

                    impacted_nodes = file_data["impacted_nodes"]

                    # DBT impacts
                    if impacted_nodes["dbt"]["nodes"]:
                        formatted_output.append(f"**🔶 dbt nodes impacted**: \n")
                        for node in impacted_nodes["dbt"]["nodes"]:
                            formatted_output.append(f"- {node['model_name']}.*{node['column_name']}*\n")

                    # Looker impacts
                    if any([
                        impacted_nodes["looker"]["views"],
                        impacted_nodes["looker"]["explores"],
                        impacted_nodes["looker"]["dashboards"]
                    ]):
                        formatted_output.append(f"**📊 Looker nodes impacted**\n")

                        if impacted_nodes["looker"]["views"]:
                            formatted_output.append("**Views:**\n")
                            for view in impacted_nodes["looker"]["views"]:
                                node_text = f"{view['model_name']}.{view['view_id']}.**{view['field_id']}**"
                                if 'url' in view and view['url']:
                                    formatted_output.append(f"- [{node_text}]({view['url']})\n")
                                else:
                                    formatted_output.append(f"- {node_text}\n")

                        if impacted_nodes["looker"]["explores"]:
                            formatted_output.append("**Explores:**\n")
                            for explore in impacted_nodes["looker"]["explores"]:
                                node_text = f"{explore['model_name']}.{explore['explore_id']}.**{explore['field_id']}**"
                                if 'url' in explore and explore['url']:
                                    formatted_output.append(f"- [{node_text}]({explore['url']})\n")
                                else:
                                    formatted_output.append(f"- {node_text}\n")

                        if impacted_nodes["looker"]["dashboards"]:
                            formatted_output.append("**Dashboards:**\n")
                            for dashboard in impacted_nodes["looker"]["dashboards"]:
                                if 'url' in dashboard and dashboard['url']:
                                    formatted_output.append(f"- [{dashboard['name']}]({dashboard['url']})\n")
                                else:
                                    formatted_output.append(f"- {dashboard['name']}\n")

                    # Tableau impacts
                    if any([
                        impacted_nodes["tableau"]["dashboards"],
                        impacted_nodes["tableau"]["datasources"],
                        impacted_nodes["tableau"]["worksheets"]
                    ]):
                        formatted_output.append(f"**📊 Tableau nodes impacted**\n")

                        if impacted_nodes["tableau"]["dashboards"]:
                            formatted_output.append("**Dashboards:**\n")
                            for dashboard in impacted_nodes["tableau"]["dashboards"]:
                                if 'url' in dashboard and dashboard['url']:
                                    formatted_output.append(f"- [{dashboard['name']}]({dashboard['url']})\n")
                                else:
                                    formatted_output.append(f"- {dashboard['name']}\n")

                        if impacted_nodes["tableau"]["datasources"]:
                            formatted_output.append("**Datasources:**\n")
                            for datasource in impacted_nodes["tableau"]["datasources"]:
                                if 'url' in datasource and datasource['url']:
                                    formatted_output.append(f"- [{datasource['name']}]({datasource['url']})\n")
                                else:
                                    formatted_output.append(f"- {datasource['name']}\n")

                        if impacted_nodes["tableau"]["worksheets"]:
                            formatted_output.append("**Worksheets:**\n")
                            for worksheet in impacted_nodes["tableau"]["worksheets"]:
                                if 'url' in worksheet and worksheet['url']:
                                    formatted_output.append(f"- [{worksheet['name']}]({worksheet['url']})\n")
                                else:
                                    formatted_output.append(f"- {worksheet['name']}\n")

                    # ThoughtSpot impacts
                    if any([
                        impacted_nodes["thoughtspot"]["answers"],
                        impacted_nodes["thoughtspot"]["liveboards"],
                        impacted_nodes["thoughtspot"]["visualizations"],
                        impacted_nodes["thoughtspot"]["worksheets"]
                    ]):
                        formatted_output.append(f"**📊 ThoughtSpot nodes impacted**\n")

                        if impacted_nodes["thoughtspot"]["answers"]:
                            formatted_output.append("**Answers:**\n")
                            for answer in impacted_nodes["thoughtspot"]["answers"]:
                                if 'url' in answer and answer['url']:
                                    formatted_output.append(f"- [{answer['name']}]({answer['url']})\n")
                                else:
                                    formatted_output.append(f"- {answer['name']}\n")

                        if impacted_nodes["thoughtspot"]["liveboards"]:
                            formatted_output.append("**Liveboards:**\n")
                            for liveboard in impacted_nodes["thoughtspot"]["liveboards"]:
                                if 'url' in liveboard and liveboard['url']:
                                    formatted_output.append(f"- [{liveboard['name']}]({liveboard['url']})\n")
                                else:
                                    formatted_output.append(f"- {liveboard['name']}\n")

                        if impacted_nodes["thoughtspot"]["visualizations"]:
                            formatted_output.append("**Visualizations:**\n")
                            for viz in impacted_nodes["thoughtspot"]["visualizations"]:
                                if 'url' in viz and viz['url']:
                                    formatted_output.append(f"- [{viz['name']}]({viz['url']})\n")
                                else:
                                    formatted_output.append(f"- {viz['name']}\n")

                        if impacted_nodes["thoughtspot"]["worksheets"]:
                            formatted_output.append("**Worksheets:**\n")
                            for worksheet in impacted_nodes["thoughtspot"]["worksheets"]:
                                if 'url' in worksheet and worksheet['url']:
                                    formatted_output.append(f"- [{worksheet['name']}]({worksheet['url']})\n")
                                else:
                                    formatted_output.append(f"- {worksheet['name']}\n")

                return '\n'.join(formatted_output)

            paradime = Paradime(
                workspace_uid=os.environ["PARADIME_WORKSPACE_UID"],
                api_secret=os.environ["PARADIME_API_SECRET"],
                api_endpoint=os.environ["PARADIME_API_ENDPOINT"]
            )

            report = paradime.lineage_diff.trigger_report_and_wait(
                bolt_run_id=int(os.environ["BOLT_RUN_ID"]),
                pull_request_number=int(os.environ["PULL_REQUEST_NUMBER"]),
                user_email=os.environ["USER_EMAIL"],
                changed_file_paths=os.environ["CHANGED_FILES"].split("\n"),
            )

            print("Lineage Diff Report Message: ", report.message)

            if report.result_json:
                # Convert JSON string to Python dict and format as markdown
                report_data = json.loads(report.result_json)
                formatted_markdown = format_json_to_markdown(report_data)
                Path(os.environ["REPORT_FILE"]).write_text(formatted_markdown)
            EOL
          - python3 generate_report.py
          - |
            if [ -f "$REPORT_FILE" ]; then
              # Create the JSON payload with properly escaped content
              FORMATTED_CONTENT=$(cat "$REPORT_FILE")
              JSON_PAYLOAD=$(jq -n \
                --arg content "$FORMATTED_CONTENT" \
                '{content:{raw:$content}}')

              # Send to Bitbucket API
              curl -X POST \
                "https://api.bitbucket.org/2.0/repositories/$BITBUCKET_REPO_FULL_NAME/pullrequests/$BITBUCKET_PR_ID/comments" \
                -H "Authorization: Bearer $BITBUCKET_ACCESS_TOKEN" \
                -H "Content-Type: application/json" \
                -d "$JSON_PAYLOAD"
            else
              echo "Report file not found, skipping comment."
            fi

pipelines:
  pull-requests:
    '**':
      - step:
          <<: *paradime-turbo-ci
      - step:
          <<: *paradime-lineage-diff-report

Generate BitBucket Access Token

You will need to create a BitBucket Access Token with pull_request:write to enable Comments to be generated when opening a Pull Request. Go to bitbucket.org and sign in:

Select your BitBucket Repository
Click “Repository Settings” in the left sidebar
In Settings, scroll to “Security” section
Select “Access tokens”
Create New Access Token
Click “Create repository access token”
Enter the following details:
- Name: Paradime
- Permissions: Select Pull requests: Write

Add the API keys and Credentials in the BitBucket Pipeline variables

Finally you need to add the API key and credentials generated in the previous step in BitBucket Pipelines as well as the BitBucket Access Token. Set the corresponding values using your credentials for the variable names:

PARADIME_WORKSPACE_UID
PARADIME_API_SECRET
PARADIME_API_ENDPOINT
BITBUCKET_ACCESS_TOKEN

Schedule Anatomy

Deploy Schedules

Managing Schedules

CI/CD

Column-Level Lineage Diff on BitBucket

Generate API keys

Generate API keys

Create a BitBucket Pipeline

Generate BitBucket Access Token

Add the API keys and Credentials in the BitBucket Pipeline variables

​Generate API keys

Generate API keys

​Create a BitBucket Pipeline

​Generate BitBucket Access Token

​Add the API keys and Credentials in the BitBucket Pipeline variables

Generate API keys

Create a BitBucket Pipeline

Generate BitBucket Access Token

Add the API keys and Credentials in the BitBucket Pipeline variables