Documentation
ドキュメントの検索
karat

+

K

APIリファレンス ↗
コードの例データセットのメタデータ操作ローカル環境

注: 以下の翻訳の正確性は検証されていません。AIPを利用して英語版の原文から機械的に翻訳されたものです。

ローカル環境

Python

データセットの行数

多数のデータセットの行数を一括で計算するにはどうすればよいですか？

このコードでは、Foundry API を使用してデータセット RID のリストに対する行数の計算を開始し、その結果を読み取ります。計算はバッチ処理できないため、コードはデータセット RID とブランチをパラメーターとして、データセットごとに1件の POST リクエストを Foundry Stats API に送信します。各 POST リクエストは計算をキューに追加するだけで、行数ではなく、解決済みのデータセットビューを返します。その後、コードはバッチエンドポイントへの1件の POST リクエストですべての結果を読み取ります。このエンドポイントはデータセット RID とブランチのペアのリストを受け取り、リクエストごとに1件のレスポンスを同じ順序で返します。非常に大きなリストは、小さなバッチに分割してください。計算は非同期で行われ、大きなデータセットでは1分以上かかる場合があるため、各データセットの status が SUCCEEDED になるまでバッチリクエストを繰り返してください。以下のコードでは、読み取りの間に1分間待機します。ステータス ABSENT、PENDING、RUNNING は行数がまだ利用可能でないことを意味し、FAILED は計算によって行数が生成されないことを意味します。返される rowCount は数値ではなくラッパーオブジェクトであり、正確に計算された件数は exactValue.value で Decimal 文字列として取得できます。

Foundry はデータセットのビルド中にも統計情報を記録しますが、この収集はベストエフォートで行われます。ビルド自体は失敗しなくても、収集がタイムアウトしたり失敗したりする場合があるため、一部のデータセットにはビルド時の行数がありません。以下のように計算を開始すると、統計情報が記録されていないデータセットの行数を生成できます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
from shutil import ExecError
from wsgiref import headers
import requests
from urllib3 import Retry
import json
import pprint
import time

'''
指定されたデータセットの RID に対して、スクリプトが行数の計算を開始します
'''

# 基本変数
base_url = "https://STACK_NAME.palantircloud.com"
branch = "master"

DATASETS_RIDS = [
    "ri.foundry.main.dataset.6d2cd3de-0052-xxxxx-c7ae2c4ab1d8"
]

headers = {
    'Authorization': 'Bearer eyg_PUT_YOUR_TOKEN_HERE_xxxx',
    'Content-Type': 'application/json'
}

# プロキシ
proxyDict = {
    # "https": "https://proxyIfNeeded:port"
}

# 再試行
retry = Retry(connect=1, backoff_factor=0.5)
adapter = requests.adapters.HTTPAdapter(max_retries=retry)
http = requests.Session()
http.mount("https://", adapter)

def trigger_row_count(dataset_rid, branch):
    response = http.post(f'{base_url}/foundry-stats/api/stats/datasets/{dataset_rid}/branches/{branch}', headers=headers,
                         proxies=proxyDict)
    raw_response = response.text
    curr_response = json.loads(raw_response)
    pprint.pprint(curr_response)

    return curr_response

def get_row_counts(dataset_rids, branch, attempts=12, delay=60):
    row_counts = {}
    pending_rids = list(dataset_rids)

    for _ in range(attempts):
        stats_requests = [{'datasetRid': pending_rid, 'branch': branch} for pending_rid in pending_rids]
        response = http.post(f'{base_url}/foundry-stats/api/stats/batch/stats', headers=headers,
                             proxies=proxyDict, data=json.dumps(stats_requests))
        if not response.ok:
            print(f"Could not read row counts: {response.text}")
            return row_counts

        # バッチエンドポイントは、リクエストの送信順に、リクエストごとに1件のレスポンスを返します。
        still_pending_rids = []
        for dataset_rid, curr_response in zip(pending_rids, json.loads(response.text)):
            status = curr_response['status']
            if status == 'SUCCEEDED':
                result = curr_response['result']
                row_counts[dataset_rid] = result[result['type']]['stats']['rowCount']
            elif status == 'FAILED':
                print(f"Row count computation failed for {dataset_rid}")
            else:
                still_pending_rids.append(dataset_rid)

        pending_rids = still_pending_rids
        if not pending_rids:
            break

        time.sleep(delay)

    for pending_rid in pending_rids:
        print(f"Row count for {pending_rid} is still not ready")

    pprint.pprint(row_counts)

    return row_counts

for curr_dataset_rid in DATASETS_RIDS :
    trigger_row_count(curr_dataset_rid, branch)

get_row_counts(DATASETS_RIDS, branch)
  • 登録日: 2024-03-26
  • タグ: export, python, metrics, metadata, local

データセット全体の列のスーパーセットを取得する

複数のデータセットに含まれるすべての列の集合を取得するにはどうすればよいですか？

このコードでは、requests ライブラリを使用して対象データセットのリストにある各データセットのスキーマを取得し、スキーマ内のフィールドを順に処理して、列のスーパーセットに含まれる各列の出現頻度を格納する辞書を作成します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
import time

from requests.packages.urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
import requests
import json
import pprint
import logging
import datetime
import collections


'''
データセット群から、列のスーパーセットを出現頻度とともに生成するスクリプトです
'''

headers = {
    'Authorization': 'Bearer eyg_PUT_YOUR_TOKEN_HERE_xxxx',
    'Content-Type': 'application/json',
}

## STACK_NAME
base_url = "STACK_NAME.palantircloud.com"
branch = "master"


target_datasets = ["ri.foundry.main.dataset.4c2ac089-xxxx-4df863eaf823"]

# プロキシ
proxyDict = {
    #"https": "https://proxyIfNeeded:port"
}

# 再試行
retry = Retry(connect=1, backoff_factor=0.5)
adapter = HTTPAdapter(max_retries=retry)
http = requests.Session()
http.mount("https://", adapter)


global_list_fields = {}

for curr_dataset in target_datasets :
    # データセットのスキーマを取得します
    print(f"Step 1. Get Schema of dataset")
    response = http.get(f'{base_url}/foundry-metadata/api/schemas/datasets/{curr_dataset}/branches/{branch}', headers=headers, proxies=proxyDict)
    print(f"Step 1. Response of getting schema of dataset")
    raw_response = response.text
    print(raw_response)
    curr_schema = json.loads(raw_response)
    list_fields = curr_schema["schema"]["fieldSchemaList"]

    for field in list_fields:
        curr_key = f"{field['name']} - {field['type']}"
        # カウンターを増やします
        global_list_fields[curr_key] = global_list_fields.get(curr_key, 0) + 1

print("Unsorted dict")
pprint.pprint(global_list_fields)

# 並べ替えます
sorted_dict = {k: v for k, v in sorted(global_list_fields.items(), key=lambda item: item[1])}
print("Sorted dict")
pprint.pprint(sorted_dict)
  • 登録日: 2024-03-26
  • タグ: python, API, metadata, code repositories, code authoring, local

指定されたリソース RID のパスを取得する

リソースの RID からそのパスを調べるにはどうすればよいですか？

このコードでは、requests ライブラリを使用して、指定された RID を含む HTTP GET リクエストを指定のホストに送信し、リソースのパスを取得します。再試行とプロキシも処理します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
from requests.adapters import HTTPAdapter
import requests
from urllib3 import Retry

'''
Script to return the path of a given Resource IDentifier (RID).
'''

# Headers
headers = {
    'Authorization': 'Bearer xxx', # Replace 'xxx' with your bearer token
    'Content-Type': 'application/json',
}

# Host
host = 'host.com:443'

# Proxies
proxyDict = {
    'https': 'http://proxy.domain.com:3333'
}

# Retries
retry = Retry(connect=1, backoff_factor=0.5)
adapter = HTTPAdapter(max_retries=retry)
http = requests.Session()
http.mount('https://', adapter)

# Enter the rid of the resource you want the path of
RESOURCE_RID = ''

# Throw an error if the reference has failed to be created
try:
    print(f'Fetching path for rid {RESOURCE_RID} ...')
    response = http.get(f'https://{host}/compass/api/resources/{RESOURCE_RID}/path-json', headers=headers, proxies=proxyDict)
    print('Completed request')
    print(f'The path is: {response.text}')
except requests.exceptions.RequestException as e:
    raise Exception(f"An error occurred in the request.\nReturning the path for the repository: {RESOURCE_RID} failed due to: {response.status_code} - {response.text}\nException: {e}")
  • 登録日: 2024-03-26
  • タグ: api, python, metadata, local