使用 DataWorks OpenAPI 進行血緣查詢 - DataWorks

本文檔旨在指導您如何使用 DataWorks OpenAPI（2024-05-18），以編程方式查詢資料表和欄位的血緣關係。我們將通過具體的 API 呼叫樣本和 SDK 代碼，協助您快速上手，滿足自動化、批量化的血緣分析需求。

什麼是資料血緣？

想象一下，你正在看一份重要的商業報表，上面顯示本季度的銷售額大幅增長。作為一個嚴謹的資料分析師或管理者，你腦海中會立刻浮現出幾個問題：

這個“銷售額”指標是如何計算出來的？
它來源於哪些原始的業務資料？是訂單表？還是支付流水表？
在從未經處理資料到最終報表的整個過程中，它經過了哪些處理步驟？（比如清洗、轉換、彙總等）
如果這個指標的資料出錯了，會影響到下遊哪些其他報表或應用？

清晰的資料血緣至關重要，它主要帶來以下核心價值：

資料溯源與問題排查
當探索資料異常或錯誤時，可以沿著血緣關係向上追溯，快速定位到是哪個環節的計算邏輯或源頭資料出了問題，極大地縮短故障排查時間。
影響分析
當需要對某個資料表結構、欄位或計算邏輯進行變更時，可以通過血緣關係向下分析，精確評估該變更會影響到哪些下遊的資料和業務報表，從而避免“牽一髮而動全身”的未知風險。
資料治理與可信度
清晰的血緣關係是資料資產管理、資料標準落地和資料品質監控的基礎。它讓資料的來龍去脈變得透明，從而增強業務方對資料的信任度。
成本最佳化與資產盤點
通過分析血緣，可以識別出那些“無人問津”（沒有下遊消費）的資料表或計算任務，為資料倉儲的成本最佳化和老舊資產的下線提供決策依據。

在 DataWorks 中，系統會自動解析並記錄各種計算任務（如MaxCompute SQL、EMR Spark等）產生的資料血緣關係。而通過 DataWorks OpenAPI，您可以以編程的方式訪問這些血緣資訊，從而將血緣分析能力整合到您自己的資料管理平台或自動化營運流程中。

準備工作：擷取實體 ID

在進行任何血緣查詢之前，您首先需要擷取目標資料（表或欄位）的唯一識別碼，即實體ID (Entity ID)。實體 ID 是調用中繼資料和血緣相關 API 的核心憑據。

您可以通過以下兩種方式擷取實體 ID：

1. 通過 DataWorks 介面擷取

對於少量、已知的表或欄位，通過介面手動複製是最快捷的方式。

擷取表的實體 ID

進入 DataWorks 的資料地圖模組。
搜尋並進入您要查詢的表詳情頁。
在左側的錶基礎信息面板中，找到實體ID並複製。

擷取欄位的實體 ID

在目標表的詳情頁，切換到血緣資訊選項卡，並選擇欄位血緣。
在欄位血緣圖譜中，點擊您關注的欄位節點。
右側會彈出該欄位的詳情面板，在面板中找到實體ID並複製。

2. 通過 API 批量擷取

當您需要批量擷取實體 ID 時，手動操作會非常繁瑣。此時，建議使用 OpenAPI 進行批量查詢：

批量擷取表ID：調用 ListTables API，詳情請參見ListTables - 查詢資料地圖資料表列表。
批量擷取欄位ID：調用 ListColumns API，詳情請參見ListColumns - 查詢資料地圖資料表欄位列表。

使用 `ListLineages` API 查詢血緣

擷取實體 ID 後，您就可以使用核心的 ListLineages API 來查詢其上下遊血緣關係了。

1. API 核心參數說明

以下是 ListLineages API 的關鍵請求參數。您可以在 OpenAPI 門戶進行線上調試。

參數	類型	描述
`SrcEntityId`	String	查詢下遊血緣時使用。傳入源頭（上遊）實體ID，API 將返回該實體的所有下遊血緣。
`DstEntityId`	String	查詢上遊血緣時使用。傳入目標（下遊）實體ID，API 將返回該實體的所有上遊血緣。
`SrcEntityName`	String	配合 `DstEntityId` 使用，用於模糊搜尋和過濾上遊實體。
`DstEntityName`	String	配合 `SrcEntityId` 使用，用於模糊搜尋和過濾下遊實體。
`NeedAttachRelationship`	Boolean	是否在響應中包含詳細的血緣關係資訊。建議設定為 `true` 以擷取完整上下文。

重要

如果同時提供SrcEntityId與DstEntityId，會返回給定上下遊血緣實體下的血緣關係。
如果SrcEntityId與DstEntityId為同一ID，會返回該實體只指向自己的血緣關係。

2. 調用樣本

假設我們有一個 MaxCompute 表，其實體ID為 maxcompute-table:::test_project::test_table。

樣本一：查詢該表的下遊血緣

要查詢這個表的所有下遊表，您需要將它作為源頭 (Source)：

SrcEntityId: maxcompute-table:::test_project::test_table
NeedAttachRelationship: true

如果只想尋找下遊表中名稱包含 "report" 的表，可以追加 DstEntityName 參數：

DstEntityName: report

樣本二：查詢該表的上遊血緣

要查詢是哪些表或任務產生了這個表，您需要將它作為目標 (Destination)：

DstEntityId: maxcompute-table:::test_project::test_table
NeedAttachRelationship: true

同樣，您也可以通過 SrcEntityName 參數來過濾上遊來源。

3. 理解 API 響應

調用 ListLineages 成功後，您會得到一個包含多條血緣關係的列表。每一條血緣關係都包含源實體、目標實體以及它們之間的關聯資訊。

單條血緣關係響應樣本 (JSON):

{
  "SrcEntity": {
    "Id": "maxcompute-table:::test_project::table_from",
    "Name": "table_from",
    "Attributes": {
      "rawEntityId": "maxcompute-table:::test_project::table_from"
    }
  },
  "DstEntity": {
    "Id": "maxcompute-table:::test_project::table_to",
    "Name": "table_to",
    "Attributes": {
      "project": "test_project",
      "region": "cn-shanghai",
      "table": "table_to"
    }
  },
  "Relationships": [
    {
      "Id": "123456789:maxcompute-table.test_project.table_from:maxcompute-table.test_project.table_to:maxcompute.SQL.76543xxx",
      "CreateTime": 1761089163548,
      "Task": {
        "Id": "76543xxx",
        "Type": "dataworks-sql",
        "Attributes": {
          "engine": "maxcompute",
          "channel": "1st",
          "taskInstanceId": "12345xxx",
          "projectId": "123456",
          "taskId": "76543xxx"
        }
      }
    }
  ]
}

如何解讀響應：

SrcEntity 和 DstEntity: 分別代表血緣的上遊和下遊實體。您可以拿到它們的 Id，並進一步調用GetTable或GetColumnAPI 來擷取更詳細的中繼資料資訊。
Relationships: 描述了 SrcEntity 和 DstEntity 是如何關聯起來的。
- Task: 描述產生這條血緣關係的任務資訊。如果任務是 DataWorks 調度任務，Task.Attributes 中會包含taskId和taskInstanceId。您可以使用這些 ID 調用GetTaskAPI 來擷取任務的詳細定義和運行狀態。

Java SDK 實戰演練

下面以 Java SDK 為例，展示如何通過代碼實現完整的血緣查詢流程。

1. 環境準備

JDK 版本: 確保您已安裝 JDK 8 或更高版本。
Maven 依賴: 在您的專案 pom.xml 檔案中添加以下依賴。請將 ${latest.version} 替換為 SDK 的最新版本號碼

<dependency>
    <groupId>com.aliyun</groupId>
    <artifactId>dataworks_public20240518</artifactId>
    <version>${latest.version}</version>
</dependency>

2. 完整程式碼範例

以下代碼示範了如何初始化用戶端、查詢指定表的上遊和下遊血緣，並列印關鍵資訊。

import java.util.List;
import java.util.Map;

import com.aliyun.dataworks_public20240518.Client;
import com.aliyun.dataworks_public20240518.models.GetTableRequest;
import com.aliyun.dataworks_public20240518.models.GetTableResponse;
import com.aliyun.dataworks_public20240518.models.LineageEntity;
import com.aliyun.dataworks_public20240518.models.LineageRelationship;
import com.aliyun.dataworks_public20240518.models.LineageTask;
import com.aliyun.dataworks_public20240518.models.ListLineagesRequest;
import com.aliyun.dataworks_public20240518.models.ListLineagesResponse;
import com.aliyun.dataworks_public20240518.models.ListLineagesResponseBody.ListLineagesResponseBodyPagingInfo;
import com.aliyun.dataworks_public20240518.models.ListLineagesResponseBody.ListLineagesResponseBodyPagingInfoLineages;
import com.aliyun.dataworks_public20240518.models.Table;
import com.aliyun.tea.TeaException;

public class LineageQuerySample {
  /**
     * <b>description</b> :
     * <p>使用憑據初始化帳號 Client</p>
     *
     * @return Client
     * @throws Exception
     */
  public static com.aliyun.dataworks_public20240518.Client createClient() throws Exception {
    com.aliyun.teaopenapi.models.Config config = new com.aliyun.teaopenapi.models.Config()
      // 您的 AccessKey ID
      .setAccessKeyId(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"))
      // 您的 AccessKey Secret
      .setAccessKeySecret(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
    // Endpoint 請參考 https://api.aliyun.com/product/dataworks-public
    config.endpoint = "dataworks.cn-hangzhou.aliyuncs.com";
    return new com.aliyun.dataworks_public20240518.Client(config);
  }

  public static void main(String[] args_) throws Exception {
    Client client = LineageQuerySample.createClient();
    // 待查詢的表實體ID，此處需要替換為待查詢的mc表實體ID
    String tableId = "maxcompute-table:::test_project::test_table";
    try {
      // 1. 查詢上遊血緣
      ListLineagesRequest listLineagesRequest = new ListLineagesRequest()
        .setDstEntityId(tableId)
        .setNeedAttachRelationship(true)
        .setPageNumber(1)
        // 預設查詢10條記錄，最大為100
        .setPageSize(10);
      // 支援對血緣上遊表名稱進行關鍵詞匹配過濾
      listLineagesRequest.setSrcEntityName("demo");
      ListLineagesResponse listLineagesResponse = client.listLineages(listLineagesRequest);
      String requestId = listLineagesResponse.getBody().getRequestId();
      System.out.println("\n查詢上遊血緣");
      // 列印requestId，輔助問題排查
      System.out.println(requestId);
      ListLineagesResponseBodyPagingInfo pagingInfo = listLineagesResponse.getBody().getPagingInfo();
      if (pagingInfo.getTotalCount() > 0 && pagingInfo.getLineages() != null) {
        for (ListLineagesResponseBodyPagingInfoLineages lineage : pagingInfo.getLineages()) {
          // 擷取單條血緣，查詢對應血緣上遊表
          LineageEntity srcEntity = lineage.getSrcEntity();
          System.out.println("============================================");
          System.out.println("ID: " + srcEntity.getId());
          System.out.println("Name: " + srcEntity.getName());
          // 擷取上遊表資訊
          Table table = getTable(client, srcEntity.getId());
          if (table != null) {
            System.out.println("Comment: " + table.getComment());
            System.out.println("Create Time: " + table.getCreateTime());
            System.out.println("Modify Time: " + table.getModifyTime());
          }
        }
      }
            // 2. 查詢下遊血緣
            listLineagesRequest = new ListLineagesRequest()
                    .setSrcEntityId(tableId)
                    .setNeedAttachRelationship(true)
                    .setPageNumber(1)
                    // 預設查詢10條記錄，最大為100
                    .setPageSize(10);
            listLineagesResponse = client.listLineages(listLineagesRequest);
            requestId = listLineagesResponse.getBody().getRequestId();
            System.out.println("\n查詢下遊血緣");
            // 列印requestId，輔助問題排查
            System.out.println(requestId);
            pagingInfo = listLineagesResponse.getBody().getPagingInfo();
            if (pagingInfo.getTotalCount() > 0 && pagingInfo.getLineages() != null) {
                for (ListLineagesResponseBodyPagingInfoLineages lineage : pagingInfo.getLineages()) {
                    // 擷取單條血緣，查詢對應血緣下遊表
                    LineageEntity dstEntity = lineage.getDstEntity();
                    System.out.println("============================================");
                    System.out.println("ID: " + dstEntity.getId());
                    System.out.println("Name: " + dstEntity.getName());
                    // 擷取下遊表資訊
                    Table table = getTable(client, dstEntity.getId());
                    if (table != null) {
                        System.out.println("Comment: " + table.getComment());
                        System.out.println("Create Time: " + table.getCreateTime());
                        System.out.println("Modify Time: " + table.getModifyTime());
                    }
                    // 解析血緣關係
                    List<LineageRelationship> relationships = lineage.getRelationships();
                    if (relationships != null) {
                        for (LineageRelationship relationship : relationships) {
                            System.out.println("\n\tRelationshipId: " + relationship.getId());
                            System.out.println("\tRelationshipCreateTime: " + relationship.getCreateTime());
                            // 解析任務詳情
                            LineageTask task = relationship.getTask();
                            Map<String, String> attributes = task.getAttributes();
                            // 對於DataWorks調度任務，可以通過attributes擷取任務ID和任務執行個體ID
                            if (attributes != null && attributes.containsKey("taskId") && attributes.containsKey("taskInstanceId")) {
                                System.out.println("\tTaskId: " + attributes.get("taskId"));
                                System.out.println("\tTaskInstanceId: " + attributes.get("taskInstanceId"));
                            }
                        }
                    }
                }
            }
        } catch (TeaException error) {
            // 此處僅做列印展示，請謹慎對待異常處理，在工程專案中切勿直接忽略異常。
            // 錯誤 message
            System.out.println(error.getMessage());
            // 診斷地址
            System.out.println(error.getData().get("Recommend"));
            com.aliyun.teautil.Common.assertAsString(error.message);
        } catch (Exception _error) {
            TeaException error = new TeaException(_error.getMessage(), _error);
            // 此處僅做列印展示，請謹慎對待異常處理，在工程專案中切勿直接忽略異常。
            // 錯誤 message
            System.out.println(error.getMessage());
            // 診斷地址
            System.out.println(error.getData().get("Recommend"));
            com.aliyun.teautil.Common.assertAsString(error.message);
        }
    }

    public static Table getTable(Client client, String tableId) {
        // 根據ID查詢表資訊
        GetTableRequest getTableRequest = new GetTableRequest()
                .setId(tableId)
                .setIncludeBusinessMetadata(true);
        try {
            GetTableResponse getTableResponse = client.getTable(getTableRequest);
            return getTableResponse.getBody().getTable();
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }
        return null;
    }
}