DataWorks:Kueri alur data menggunakan DataWorks OpenAPI

1. Dapatkan entity ID dari Konsol

Untuk sejumlah kecil tabel atau bidang yang sudah diketahui, metode tercepat adalah menyalin ID secara manual dari Konsol.

2. Dapatkan entity ID secara massal melalui API

Jika Anda perlu memperoleh banyak entity ID, operasi manual bisa menjadi merepotkan. Dalam kasus ini, gunakan OpenAPI untuk melakukan kueri massal:

• Dapatkan ID tabel secara massal: Panggil API ListTables. Untuk informasi lebih lanjut, lihat Kueri daftar tabel data di Data Map.

• Dapatkan ID bidang secara massal: Panggil API ListColumns. Untuk informasi lebih lanjut, lihat Kueri daftar bidang dalam tabel data di Data Map.

1. Parameter API utama

Tabel berikut menjelaskan parameter permintaan utama API ListLineages. Anda dapat melakukan debug API secara daring di OpenAPI Portal.

2. Contoh pemanggilan

Asumsikan Anda memiliki tabel MaxCompute dengan entity ID-nya adalah maxcompute-table:::test_project::test_table.

3. Respons API

Pemanggilan sukses ke ListLineages mengembalikan daftar hubungan alur data. Setiap hubungan alur data berisi entitas sumber, entitas tujuan, dan informasi tentang asosiasinya.

Contoh tanggapan untuk hubungan lineage tunggal (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"
        }
      }
    }
  ]
}

Cara menafsirkan respons:

• SrcEntity dan DstEntity: Mewakili entitas hulu dan hilir dari alur data, secara berurutan. Anda dapat menggunakan Id-nya untuk memanggil API GetTable atau GetColumn untuk mendapatkan metadata lebih detail.

• Relationships: Menjelaskan bagaimana SrcEntity dan DstEntity diasosiasikan.

  • Task: Menjelaskan task yang membuat hubungan alur data ini. Jika task tersebut merupakan task terjadwal DataWorks, Task.Attributes berisi taskId dan taskInstanceId. Anda dapat menggunakan ID-ID ini untuk memanggil API GetTask guna mendapatkan definisi task dan status eksekusi secara detail.

1. Prasyarat

• Versi JDK: JDK 8 atau yang lebih baru.

• Dependensi Maven: Tambahkan dependensi berikut ke file pom.xml proyek Anda. Ganti ${latest.version} dengan nomor versi terbaru SDK.

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

2. Contoh kode lengkap

Kode berikut menunjukkan cara menginisialisasi client, mengkueri alur data hulu dan hilir dari tabel tertentu, serta mencetak informasi penting.

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 {
  /**
     * description :
     * <p>Menginisialisasi client menggunakan kredensial.</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()
      // ID AccessKey Anda.
      .setAccessKeyId(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"))
      // Rahasia AccessKey Anda.
      .setAccessKeySecret(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
    // Untuk endpoint, lihat https://api.alibabacloud.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 entitas tabel yang akan dikueri. Ganti dengan ID entitas tabel MaxCompute Anda.
    String tableId = "maxcompute-table:::test_project::test_table";
    try {
      // 1. Kueri alur data hulu.
      ListLineagesRequest listLineagesRequest = new ListLineagesRequest()
        .setDstEntityId(tableId)
        .setNeedAttachRelationship(true)
        .setPageNumber(1)
        // Secara default, 10 catatan dikembalikan. Nilai maksimum adalah 100.
        .setPageSize(10);
      // Mendukung pencocokan fuzzy berbasis kata kunci dan penyaringan nama tabel hulu.
      listLineagesRequest.setSrcEntityName("demo");
      ListLineagesResponse listLineagesResponse = client.listLineages(listLineagesRequest);
      String requestId = listLineagesResponse.getBody().getRequestId();
      System.out.println("\nKueri alur data hulu");
      // Cetak ID permintaan untuk troubleshooting.
      System.out.println(requestId);
      ListLineagesResponseBodyPagingInfo pagingInfo = listLineagesResponse.getBody().getPagingInfo();
      if (pagingInfo.getTotalCount() > 0 && pagingInfo.getLineages() != null) {
        for (ListLineagesResponseBodyPagingInfoLineages lineage : pagingInfo.getLineages()) {
          // Dapatkan satu hubungan alur data dan kueri tabel hulu yang sesuai.
          LineageEntity srcEntity = lineage.getSrcEntity();
          System.out.println("============================================");
          System.out.println("ID: " + srcEntity.getId());
          System.out.println("Nama: " + srcEntity.getName());
          // Dapatkan informasi tabel hulu.
          Table table = getTable(client, srcEntity.getId());
          if (table != null) {
            System.out.println("Komentar: " + table.getComment());
            System.out.println("Waktu Pembuatan: " + table.getCreateTime());
            System.out.println("Waktu Modifikasi: " + table.getModifyTime());
          }
        }
      }
            // 2. Kueri alur data hilir.
            listLineagesRequest = new ListLineagesRequest()
                    .setSrcEntityId(tableId)
                    .setNeedAttachRelationship(true)
                    .setPageNumber(1)
                    // Secara default, 10 catatan dikembalikan. Nilai maksimum adalah 100.
                    .setPageSize(10);
            listLineagesResponse = client.listLineages(listLineagesRequest);
            requestId = listLineagesResponse.getBody().getRequestId();
            System.out.println("\nKueri alur data hilir");
            // Cetak ID permintaan untuk troubleshooting.
            System.out.println(requestId);
            pagingInfo = listLineagesResponse.getBody().getPagingInfo();
            if (pagingInfo.getTotalCount() > 0 && pagingInfo.getLineages() != null) {
                for (ListLineagesResponseBodyPagingInfoLineages lineage : pagingInfo.getLineages()) {
                    // Dapatkan satu hubungan alur data dan kueri tabel hilir yang sesuai.
                    LineageEntity dstEntity = lineage.getDstEntity();
                    System.out.println("============================================");
                    System.out.println("ID: " + dstEntity.getId());
                    System.out.println("Nama: " + dstEntity.getName());
                    // Dapatkan informasi tabel hilir.
                    Table table = getTable(client, dstEntity.getId());
                    if (table != null) {
                        System.out.println("Komentar: " + table.getComment());
                        System.out.println("Waktu Pembuatan: " + table.getCreateTime());
                        System.out.println("Waktu Modifikasi: " + table.getModifyTime());
                    }
                    // Urai hubungan alur data.
                    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());
                            // Urai detail task.
                            LineageTask task = relationship.getTask();
                            Map<String, String> attributes = task.getAttributes();
                            // Untuk task terjadwal DataWorks, Anda dapat memperoleh ID task dan ID instans task dari atribut.
                            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) {
            // Di lingkungan produksi, implementasikan penanganan exception yang kuat.
            // Pesan error
            System.out.println(error.getMessage());
            // URL diagnosa
            System.out.println(error.getData().get("Recommend"));
            com.aliyun.teautil.Common.assertAsString(error.message);
        } catch (Exception _error) {
            TeaException error = new TeaException(_error.getMessage(), _error);
            // Di lingkungan produksi, implementasikan penanganan exception yang kuat.
            // Pesan error
            System.out.println(error.getMessage());
            // URL diagnosa
            System.out.println(error.getData().get("Recommend"));
            com.aliyun.teautil.Common.assertAsString(error.message);
        }
    }
    public static Table getTable(Client client, String tableId) {
        // Kueri informasi tabel berdasarkan 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;
    }
}

1. Prasyarat

• Versi Python: Python 3.6 atau yang lebih baru.

• Instal SDK: Instal DataWorks Python SDK menggunakan pip. Ganti ${latest.version} dengan nomor versi terbaru SDK.

pip install alibabacloud_dataworks_public20240518==${latest.version}

2. Contoh kode lengkap

Kode berikut menunjukkan cara menginisialisasi client, mengkueri alur data hulu dan hilir dari tabel tertentu, serta mencetak informasi penting.

# -*- coding: utf-8 -*-
import os
import sys
from alibabacloud_dataworks_public20240518.client import Client as dataworks_public20240518Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_dataworks_public20240518 import models as dataworks_public_20240518_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient


class LineageQuerySample:

    @staticmethod
    def create_client():
        """Menginisialisasi client menggunakan AccessKey Anda."""
        config = open_api_models.Config(
            # ID AccessKey Anda.
            access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID'),
            # Rahasia AccessKey Anda.
            access_key_secret=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
        )
        # Untuk endpoint, lihat https://api.alibabacloud.com/product/dataworks-public.
        config.endpoint = 'dataworks.cn-hangzhou.aliyuncs.com'
        return dataworks_public20240518Client(config)

    @staticmethod
    def get_table(client, table_id):
        """Mendapatkan informasi tabel berdasarkan entity ID."""
        get_table_request = dataworks_public_20240518_models.GetTableRequest(
            id=table_id,
            include_business_metadata=True
        )
        try:
            response = client.get_table(get_table_request)
            return response.body.table
        except Exception as e:
            print(e)
            return None

    @staticmethod
    def main():
        client = LineageQuerySample.create_client()
        # ID entitas tabel yang akan dikueri. Ganti dengan ID entitas tabel MaxCompute Anda.
        table_id = 'maxcompute-table:::test_project::test_table'
        runtime = util_models.RuntimeOptions()

        try:
            # 1. Kueri alur data hulu.
            upstream_request = dataworks_public_20240518_models.ListLineagesRequest(
                dst_entity_id=table_id,
                need_attach_relationship=True,
                page_number=1,
                # Secara default, 10 catatan dikembalikan. Nilai maksimum adalah 100.
                page_size=10,
                # Mendukung pencocokan fuzzy berbasis kata kunci dan penyaringan nama tabel hulu.
                src_entity_name='demo'
            )
            upstream_response = client.list_lineages_with_options(upstream_request, runtime)
            print('\nKueri alur data hulu')
            print(upstream_response.body.request_id)
            paging_info = upstream_response.body.paging_info
            if paging_info.total_count > 0 and paging_info.lineages:
                for lineage in paging_info.lineages:
                    src_entity = lineage.src_entity
                    print('============================================')
                    print(f'ID: {src_entity.id}')
                    print(f'Nama: {src_entity.name}')
                    table = LineageQuerySample.get_table(client, src_entity.id)
                    if table:
                        print(f'Komentar: {table.comment}')
                        print(f'Waktu Pembuatan: {table.create_time}')
                        print(f'Waktu Modifikasi: {table.modify_time}')

            # 2. Kueri alur data hilir.
            downstream_request = dataworks_public_20240518_models.ListLineagesRequest(
                src_entity_id=table_id,
                need_attach_relationship=True,
                page_number=1,
                page_size=10
            )
            downstream_response = client.list_lineages_with_options(downstream_request, runtime)
            print('\nKueri alur data hilir')
            print(downstream_response.body.request_id)
            paging_info = downstream_response.body.paging_info
            if paging_info.total_count > 0 and paging_info.lineages:
                for lineage in paging_info.lineages:
                    dst_entity = lineage.dst_entity
                    print('============================================')
                    print(f'ID: {dst_entity.id}')
                    print(f'Nama: {dst_entity.name}')
                    table = LineageQuerySample.get_table(client, dst_entity.id)
                    if table:
                        print(f'Komentar: {table.comment}')
                        print(f'Waktu Pembuatan: {table.create_time}')
                        print(f'Waktu Modifikasi: {table.modify_time}')
                    # Urai hubungan alur data.
                    if lineage.relationships:
                        for relationship in lineage.relationships:
                            print(f'\n\tRelationshipId: {relationship.id}')
                            print(f'\tRelationshipCreateTime: {relationship.create_time}')
                            task = relationship.task
                            attributes = task.attributes
                            if attributes and 'taskId' in attributes and 'taskInstanceId' in attributes:
                                print(f'\tTaskId: {attributes["taskId"]}')
                                print(f'\tTaskInstanceId: {attributes["taskInstanceId"]}')

        except Exception as error:
            # Di lingkungan produksi, implementasikan penanganan exception yang kuat.
            print(error)
            UtilClient.assert_as_string(str(error))


if __name__ == '__main__':
    LineageQuerySample.main()