Compartilhar via

Implantar uma casa de eventos usando APIs de malha

Você pode automatizar totalmente a implantação de suas Eventhouses com bancos de dados KQL usando APIs. As APIs de malha permitem criar, atualizar e excluir itens em seu espaço de trabalho. Você pode gerenciar suas Eventhouses e Bancos de Dados executando ações como criar tabelas e alterar políticas usando um dos seguintes métodos:

  • Fabric API com definição: Você pode especificar um script de esquema de base de dados como parte da definição de base de dados KQL para configurar a sua base de dados.
  • Kusto API: Você pode usar a API Kusto para executar comandos de gerenciamento de para configurar seu banco de dados.

Neste artigo, você aprende a:

  • Configurar o ambiente
  • Criar uma casa de eventos
  • Criar um banco de dados e um esquema KQL
  • Monitorizar a operação até à conclusão

Pré-requisitos

Escolha o método certo

Ao escolher o método certo para gerenciar sua Eventhouse e KQL Database, considere estes pontos:

  • Fabric API com definição: Use este método se quiser definir o esquema da sua base de dados como parte da sua definição. Esse método é útil quando você deseja definir o esquema do seu banco de dados usando uma única API consistente para toda a implantação.
  • Kusto API: Use este método se quiser executar comandos de gerenciamento para configurar seu banco de dados. Esse método é útil quando você deseja executar comandos de gerenciamento para configurar seu banco de dados.

Configurar o ambiente

Para este artigo, utiliza os blocos de notas do Fabric para executar trechos de código em python. Use o pacote sempy.fabric em pacote Python de link semântico para fazer chamadas de API usando as suas credenciais. As chamadas de API e a carga útil são idênticas, independentemente da ferramenta que você usa.

Configurando seu ambiente:

  1. Navegue até um bloco de anotações existente ou crie um novo.

  2. Em uma célula de código, insira o código para importar os pacotes:

    !pip install semantic-link --q

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

  3. Configure seu cliente para fazer as chamadas de API e defina uma variável para o ID do espaço de trabalho e um UUID para garantir que os nomes sejam exclusivos:

    client = fabric.FabricRestClient()
workspace_id = 'aaaabbbb-0000-cccc-1111-dddd2222eeee'
uuid = uuid.uuid4()

Criar uma casa de eventos

  1. Adicione uma variável para o nome da casa de eventos.

    eventhouse_name = f"{'SampleEventhouse'}_{uuid}"

  2. Use a API Fabric Create Eventhouse para criar um novo espaço de eventos. Defina o ID da casa de eventos em uma variável:

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

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

Criar um banco de dados e um esquema KQL

O API Fabric Create KQL Database utiliza definições de item para propriedades de banco de dados e esquemas que requerem cadeias de caracteres base64. As propriedades definem as políticas de retenção no nível do banco de dados e o script de esquema de banco de dados contém os comandos a serem executados para criar entidades de banco de dados.

Criar a definição de propriedades do banco de dados

Crie a cadeia de caracteres base64 para as propriedades do banco de dados. As propriedades do banco de dados definem as políticas de retenção no nível do banco de dados. Use a definição como parte da chamada de API de criação de banco de dados para criar um novo banco de dados KQL.

  1. Adicione variáveis para configurar o banco de dados KQL.

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

  2. Crie uma cadeia de caracteres base64 para as propriedades do banco de dados:

    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')

Criar a definição de esquema de banco de dados

Crie a cadeia de caracteres base64 para o esquema de banco de dados. O script de esquema de banco de dados contém os comandos a serem executados para criar entidades de banco de dados. Use a definição como parte da chamada de API de criação de banco de dados para criar um novo banco de dados KQL.

Crie uma cadeia de caracteres base64 para o esquema de banco de dados:

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')

Executar a API de criação de banco de dados

Use a API Fabric Create KQL Database para criar um novo banco de dados KQL com as políticas de retenção e o esquema definidos.

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)

Monitorizar a operação até à conclusão

Criar um item com uma definição é uma operação de longa execução que é executada de forma assíncrona. Você pode monitorizar a operação usando status_code e localização no objeto de resposta a partir da chamada Create KQL Database API, da seguinte maneira:

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}")

Conteúdo relacionado