패브릭 API를 사용하여 이벤트 하우스 배포

2025-03-24

API를 사용하여 KQL 데이터베이스를 사용하여 Eventhouses 배포를 완전히 자동화할 수 있습니다. 패브릭 API를 사용하면 작업 영역 내에서 항목을 만들고, 업데이트하고, 삭제할 수 있습니다. 다음 방법 중 하나를 사용하여 테이블 만들기 및 정책 변경과 같은 작업을 수행하여 Eventhouse 및 데이터베이스를 관리할 수 있습니다.

데이터베이스를 구성하기 위해 정의 Fabric API의 일부로 KQL 데이터베이스 정의에 데이터베이스 스키마 스크립트를 지정할 수 있습니다.
Kusto API: Kusto API를 사용하여 관리 명령 실행하여 데이터베이스를 구성할 수 있습니다.

이 문서에서는 다음을 알아봅니다.

환경 설정
이벤트 하우스 만들기
KQL 데이터베이스 및 스키마 만들기
완료를 위한 작업 모니터링

필수 구성 요소

Microsoft Fabric가 활성화된 용량을 보유한 작업 공간.
작업 영역 ID입니다. 자세한 내용은 작업 영역 ID식별을 참조하세요.

올바른 메서드 선택

Eventhouse 및 KQL 데이터베이스를 관리하는 올바른 방법을 선택할 때 다음 사항을 고려합니다.

정의 있는Fabric API: 데이터베이스의 스키마를 데이터베이스 정의의 일부로 정의하려면 이 메서드를 사용합니다. 이 메서드는 전체 배포에 대해 일관된 단일 API를 사용하여 데이터베이스의 스키마를 정의하려는 경우에 유용합니다.
Kusto API: 관리 명령을 실행하여 데이터베이스를 구성하려면 이 메서드를 사용합니다. 이 메서드는 관리 명령을 실행하여 데이터베이스를 구성하려는 경우에 유용합니다.

환경 설정

이 문서에서는 Fabric Notebook을 사용하여 python 코드 조각실행합니다. 의미 체계 링크 Python 패키지의 sempy.fabric 패키지를 사용하여 자격 증명을 사용하여 API를 호출합니다. API 호출 및 페이로드는 사용하는 도구에 관계없이 동일합니다.

환경 설정:

기존 전자 필기장으로 이동하거나 새 전자 필기장을 만듭니다.

코드 셀에서 패키지를 가져오는 코드를 입력합니다.

Fabric API 정의
Kusto API

!pip install semantic-link --q

import sempy.fabric as fabric
import time
import uuid
import base64
import json

!pip install semantic-link --q

import sempy.fabric as fabric
import time
import uuid

API를 호출하도록 클라이언트를 설정하고, 이름이 고유하도록 작업 영역 ID 및 UUID에 대한 변수를 설정합니다.
```
client = fabric.FabricRestClient()
workspace_id = 'aaaabbbb-0000-cccc-1111-dddd2222eeee'
uuid = uuid.uuid4()
```

이벤트 하우스 만들기

이벤트 하우스 이름에 대한 변수를 추가합니다.
```
eventhouse_name = f"{'SampleEventhouse'}_{uuid}"
```

Fabric Create Eventhouse API 사용하여 새 Eventhouse를 만듭니다. 변수에서 Eventhouse ID를 설정합니다.

url = f"v1/workspaces/{workspace_id}/eventhouses"
payload = {
  "displayName": f"{eventhouse_name}"
}

response = client.post(url, json=payload)
eventhouse_id = response.json()['id']

KQL 데이터베이스 및 스키마 만들기

Fabric API 정의
Kusto API

Fabric Create KQL 데이터베이스 API는 base64 문자열을 필요로 하는 데이터베이스 속성 및 스키마를 위해 항목 정의를 사용합니다. 데이터베이스 수준 보존 정책을 설정하는 속성 및 데이터베이스 스키마 스크립트에는 데이터베이스 엔터티를 만들기 위해 실행할 명령이 포함되어 있습니다.

데이터베이스 속성 정의 만들기

데이터베이스 속성에 대한 base64 문자열을 만듭니다. 데이터베이스 속성은 데이터베이스 수준 보존 정책을 설정합니다. 정의를 데이터베이스 만들기 API 호출의 일부로 사용하여 새 KQL 데이터베이스를 만듭니다.

KQL 데이터베이스를 구성하기 위한 변수를 추가합니다.

database_name = f"{'SampleDatabase'}_{uuid}"
database_cache = "3d"
database_storage = "30d"

데이터베이스 속성에 대한 base64 문자열을 만듭니다.

database_properties = {
  "databaseType": "ReadWrite",
  "parentEventhouseItemId": f"{eventhouse_id}",
  "oneLakeCachingPeriod": f"{database_cache}",
  "oneLakeStandardStoragePeriod": f"{database_storage}"
}
database_properties = json.dumps(database_properties)

database_properties_string = database_properties.encode('utf-8')
database_properties_bytes = base64.b64encode(database_properties_string)
database_properties_string = database_properties_bytes.decode('utf-8')

데이터베이스 스키마 정의 만들기

데이터베이스 스키마에 대한 base64 문자열을 만듭니다. 데이터베이스 스키마 스크립트에는 데이터베이스 엔터티를 만들기 위해 실행할 명령이 포함되어 있습니다. 정의를 데이터베이스 만들기 API 호출의 일부로 사용하여 새 KQL 데이터베이스를 만듭니다.

데이터베이스 스키마에 대한 base64 문자열을 만듭니다.

database_schema=""".create-merge table T(a:string, b:string)
.alter table T policy retention @'{"SoftDeletePeriod":"10.00:00:00","Recoverability":"Enabled"}'
.alter table T policy caching hot = 3d
"""

database_schema_string = database_schema.encode('utf-8')
database_schema_bytes = base64.b64encode(database_schema_string)
database_schema_string = database_schema_bytes.decode('utf-8')

데이터베이스 만들기 API 실행

Fabric Create KQL Database API 사용하여 정의한 보존 정책 및 스키마를 사용하여 새 KQL 데이터베이스를 만듭니다.

url = f"v1/workspaces/{workspace_id}/kqlDatabases"

payload = {
  "displayName": f"{database_name}",
  "definition": {
    "parts": [
      {
        "path": "DatabaseProperties.json",
        "payload": f"{database_properties_string}",
        "payloadType": "InlineBase64"
      },
      {
        "path": "DatabaseSchema.kql",
        "payload": f"{database_schema_string}",
        "payloadType": "InlineBase64"
      }
    ]
  }
}

response = client.post(url, json=payload)

완료를 위한 작업 모니터링

정의를 사용하여 항목을 만드는 것은 비동기적으로 실행되는 장기 실행 작업입니다. 다음과 같이 KQL 데이터베이스 만들기 API 호출에서 응답 개체의 status_code 및 위치 정보를 사용하여 작업을 모니터링할 수 있습니다.

print(f"Create request status code: {response.status_code}")
print(response.headers['Location'])
async_result_polling_url = response.headers['Location']

while True:
  async_response = client.get(async_result_polling_url)
  async_status = async_response.json().get('status').lower()
  print(f"Long running operation status: {async_status}")
  if async_status != 'running':
    break

  time.sleep(3)

print(f"Long running operation reached terminal state: '{async_status}'")

if async_status == 'succeeded':
  print("The operation completed successfully.")
  final_result_url= async_response.headers['Location']
  final_result = client.get(final_result_url)
  print(f"Final result: {final_result.json()}")
elif async_status == 'failed':
  print("The operation failed.")
else:
  print(f"The operation is in an unexpected state: {status}")

이전에 만든 eventhouse에서 KQL 데이터베이스 및 스키마를 만듭니다.

KQL 데이터베이스 만들기

KQL 데이터베이스 및 데이터베이스 수준 보존 정책을 구성하기 위한 변수를 추가합니다.
```
database_name = f"{'SampleDatabase'}_{uuid}"
database_cache = "3d"
database_storage = "30d"
```

Fabric KQL 데이터베이스 API를 사용하여 이 이벤트하우스에 새 데이터베이스를 추가합니다.

url = f"v1/workspaces/{workspace_id}/kqlDatabases"

payload = {
  "displayName": f"{database_name}",
  "creationPayload": {
    "databaseType": "ReadWrite",
    "parentEventhouseItemId": f"{eventhouseId}"
  }
}

response = client.post(url, json=payload)

테이블 만들기

Fabric Get Eventhouse API를 사용하여 이벤트하우스의 쿼리 URI를 가져옵니다.

url = f"v1/workspaces/{workspace_id}/eventhouses/{eventhouseId}"

response = client.get(url)

query_uri = response.json()['properties']['queryServiceUri']

Fabric API가 Kusto 작업을 직접 지원하지 않으므로 Kusto API 호출을 위해 요청 모듈을 가져옵니다. 그런 다음, mssparkutils 패키지를 사용하여 인증을 위한 토큰 문자열을 가져옵니다.
```
import requests
token_string = mssparkutils.credentials.getToken(f"{query_uri}")
```

Kusto 관리 API 사용하여 테이블을 만들고, 캐시 정책을 설정하고, 보존 정책을 설정합니다.

url = f"{query_uri}/v1/rest/mgmt"

payload = {
  "csl": '.execute database script with (ContinueOnErrors=true) <| .create-merge table T(a:string, b:string); .alter-merge table T policy retention softdelete = 10d; .alter table T policy caching hot = 3d',
  "db": f"{database_name}"
}

header = {'Content-Type':'application/json','Authorization': f'Bearer {token_string}'}

response = requests.post(url, json=payload, headers=header)

Eventhouse 항목 작업
KQL 데이터베이스 항목 작업
KQL 데이터베이스 정의
Kusto API 개요