Criar um conector personalizado

info

Beta

Este recurso está em Beta. Os administradores do Workspace podem controlar o acesso a este recurso na página Pré-visualizações . Consulte Gerenciar prévias do Databricks.

Os conectores personalizados permitem ingerir dados de uma fonte que o Lakeflow Connect não suporta com um conector gerenciado. Você constrói e testa seu conector, depois o implanta e coloca em execução em seu próprio Workspace do Databricks. Você não precisa registrá-lo na comunidade ou contribuir com qualquer repository compartilhado para usá-lo.

Desenvolva seu conector usando as ferramentas e os padrões no Lakeflow Comunidade Connectors repository no GitHub. O repository inclui ferramentas de desenvolvimento com tecnologia de AI para auxiliar em cada fase, incluindo pesquisa de fontes, configuração de autenticação, implementação e testes. O uso dessas ferramentas não torna seu conector um conector de comunidade. O repository fornece a estrutura e os exemplos, e seu conector permanece em seu workspace, a menos que você escolha contribuir com ele.

Se você quiser compartilhar seu conector com outros usuários mais tarde, você pode, opcionalmente, contribuí-lo para a comunidade. Para usar um conector da comunidade existente, consulte Conectores da comunidade no Lakeflow Connect.

Requisitos

Antes de começar, certifique-se de ter:

Python 3.10 ou acima
Um espaço de trabalho do Databricks com o Unity Catalog habilitado
Credenciais de API para a fonte à qual você deseja se conectar
Git instalado localmente

Configurar o repository

Clone o repository Lakeflow comunidade Connectors e instale as dependências de desenvolvimento.

Clone o repository:

Bash
git clone https://github.com/databrickslabs/lakeflow-community-connectors.git
cd lakeflow-community-connectors

Crie um ambiente virtual e instale as dependências:

Bash
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

Revise as implementações de conector existentes em src/databricks/labs/community_connector/sources/, então comece a desenvolver seu conector em um novo diretório sob esse caminho. Siga os comandos e habilidades de desenvolvimento assistido por AI do repository. Para o fluxo de trabalho recomendado, use:
Text
```
/develop-connector <your-source>
/validate-connector <your-source>
```

Implementar a interface `LakeflowConnect`

Cada conector implementa a interface LakeflowConnect, que define como seu conector autentica, descobre tabelas, retorna esquemas e lê dados.

Python
class LakeflowConnect:
    def __init__(self, options: dict[str, str]) -> None:
        """Initialize with connection parameters"""

    def list_tables(self) -> list[str]:
        """Return names of all tables supported by this connector."""

    def get_table_schema(self, table_name: str, table_options: dict[str, str]) -> StructType:
        """Return the Spark schema for a table."""

    def read_table_metadata(self, table_name: str, table_options: dict[str, str]) -> dict:
        """Return metadata: primary_keys, cursor_field, ingestion_type
        (snapshot|cdc|cdc_with_deletes|append)."""

    def read_table(self, table_name: str, start_offset: dict,
                   table_options: dict[str, str]) -> (Iterator[dict], dict):
        """Yield records as JSON dicts and return the next offset
        for incremental reads."""

    def read_table_deletes(self, table_name: str, start_offset: dict,
                           table_options: dict[str, str]) -> (Iterator[dict], dict):
        """Optional: Only required if ingestion_type is 'cdc_with_deletes'."""

Descrições de métodos

Método	Descrição
`__init__`	Recebe os parâmetros de conexão como um dicionário e inicializa o cliente da API para sua origem.
`list_tables`	Retorna os nomes de todas as tabelas (ou Endpoint de API) que seu conector expõe. O Databricks usa esta lista para preencher a IU de seleção de tabelas.
`get_table_schema`	Retorna um `StructType` do Spark que descreve o esquema da tabela fornecida. Chamado antes da primeira execução do pipeline e em cada execução quando a evolução do esquema está habilitada.
`read_table_metadata`	Retorna um dicionário com `primary_keys`, `cursor_field` e `ingestion_type`. O `ingestion_type` deve ser um dos `snapshot`, `cdc`, `cdc_with_deletes` ou `append`.
`read_table`	Produz registros como dicionários Python e retorna o próximo deslocamento para leituras incrementais. Na primeira execução, `start_offset` está vazio. Em execuções subsequentes, ele contém o deslocamento retornado pela execução anterior.
`read_table_deletes`	Opcional. Implemente este método apenas se `ingestion_type` for `cdc_with_deletes`. Produz chaves de registro excluídas e retorna o próximo offset.

Método	Descrição
`__init__`	Recebe os parâmetros de conexão como um dicionário e inicializa o cliente da API para sua origem.
`list_tables`	Retorna os nomes de todas as tabelas (ou Endpoint de API) que seu conector expõe. O Databricks usa esta lista para preencher a IU de seleção de tabelas.
`get_table_schema`	Retorna um `StructType` do Spark que descreve o esquema da tabela fornecida. Chamado antes da primeira execução do pipeline e em cada execução quando a evolução do esquema está habilitada.
`read_table_metadata`	Retorna um dicionário com `primary_keys`, `cursor_field` e `ingestion_type`. O `ingestion_type` deve ser um dos `snapshot`, `cdc`, `cdc_with_deletes` ou `append`.
`read_table`	Produz registros como dicionários Python e retorna o próximo deslocamento para leituras incrementais. Na primeira execução, `start_offset` está vazio. Em execuções subsequentes, ele contém o deslocamento retornado pela execução anterior.
`read_table_deletes`	Opcional. Implemente este método apenas se `ingestion_type` for `cdc_with_deletes`. Produz chaves de registro excluídas e retorna o próximo offset.

Desenvolva seu conector

Siga os passos para construir e validar um novo conector:

Pesquisar a API da fonte : estude as especificações da API da fonte, os mecanismos de autenticação, os limites de taxa e os esquemas de dados disponíveis. Identifique quais tabelas ou Endpoint expor.
Configurar a autenticação : gere a especificação da conexão, configure as credenciais para a fonte e verifique a conectividade a partir do seu ambiente de desenvolvimento.
Implemente o conector : codifique todos os métodos de interface LakeflowConnect necessários para conectar-se à API de origem e retornar dados no formato esperado.
Testar e iterar : faça a execução dos conjuntos de testes padrão em um sistema de fonte real e corrija quaisquer problemas. Consulte Testar seu conector para obter detalhes.
Documente o conector : escreva um README.md voltado para o usuário e gere o arquivo YAML de especificação do conector que descreve os parâmetros configuráveis do conector.
Construir o artefato de implantação : faça a execução do script de build para produzir o artefato de arquivo único que pode ser implantado em um Workspace.

Testar seu conector

O repository fornece várias abordagens de teste:

Conjunto de testes genérico (obrigatório)

Conecta-se a uma origem real usando suas credenciais fornecidas para verificar a funcionalidade de ponta a ponta, incluindo autenticação, descoberta de esquema e leituras de dados.

Bash
python -m pytest tests/generic/ --connector <your-source> --credentials credentials.json

Teste de write-back (recomendado)

Executa ciclos de gravação-leitura-verificação para validar leituras e exclusões incrementais. Isso confirma que seu acompanhamento de offset e a lógica de CDC funcionam corretamente.

Bash
python -m pytest tests/writeback/ --connector <your-source> --credentials credentials.json

Testes de unidade

Escreva testes unitários para qualquer lógica personalizada complexa em seu conector, como tratamento de paginação, coerção de tipo ou recuperação de erros.

Construir o artefato de implantação

Depois que seu conector passar pelos conjuntos de testes, faça a execução do script de merge para gerar um artefato de implantação de arquivo único. O pipeline usa este arquivo em Runtime em vez do repository completo.

Bash
python tools/scripts/merge_python_source.py --connector <your-source>

Isso produz um arquivo Python independente em dist/<your-source>/ que inclui todo o código e as dependências do conector.

Criar um pipeline de ingestão

Implante e execute seu conector em seu próprio workspace do Databricks:

Na barra lateral do seu workspace do Databricks, clique em +Novo > Adicionar ou fazer upload de dados e, em seguida, escolha a opção para adicionar um conector personalizado.
Para Nome da fonte , insira o nome do seu conector.
Para o URL do repository do GitHub , insira o URL do repository do GitHub que hospeda o código-fonte do seu conector.
Clique em Adicionar conector .
Clique em + Criar conexão ou selecione uma conexão existente e, em seguida, clique em Próximo .
Para Nome do pipeline , insira um nome para o pipeline.
Para Local do log de eventos , insira um nome de catálogo e um nome de esquema. O Databricks armazena o log de eventos do pipeline aqui. As tabelas ingeridas também são gravadas aqui por default.
Para Caminho raiz , insira o caminho do seu workspace (por exemplo, /Workspace/Users/<your-email>/connectors). A Databricks clona e armazena o código-fonte do conector aqui.
Clique em Criar pipeline .

No editor de pipeline, abra ingest.py e modifique o campo objects para incluir as tabelas que você deseja ingerir. Por exemplo:

Python
from databricks.labs.community_connector.pipeline import ingest

pipeline_spec = {
    "connection_name": "my_connector_connection",  # Required: UC connection name
    "objects": [
        {"table": {"source_table": "my_table"}},
    ],
}

ingest(spark, pipeline_spec)

Execução o pipeline manualmente ou programe-o.

Opções de configuração do pipeline

Você pode configurar as seguintes opções em ingest.py:

Opção	Descrição
`connection_name`	Obrigatório. O nome da conexão que armazena as credenciais de autenticação para a fonte.
`objects`	Obrigatório. Uma lista de tabelas para ingestão. Cada entrada tem o formato `{"table": {"source_table": "..."}}`. Você também pode especificar um `destination_table` opcional dentro do objeto `table`.
`destination_catalog`	O catálogo onde as tabelas ingeridas são gravadas. Adota como default o catálogo definido durante a criação do pipeline.
`destination_schema`	O esquema em que as tabelas ingeridas são gravadas. Adota como default o esquema definido durante a criação do pipeline.
`scd_type`	A estratégia de dimensões que mudam lentamente (SCD): `SCD_TYPE_1`, `SCD_TYPE_2` ou `APPEND_ONLY`. O default é `SCD_TYPE_1`.
`primary_keys`	Substituir as chaves primárias default de uma tabela. Forneça uma lista de nomes de colunas.

Opção	Descrição
`connection_name`	Obrigatório. O nome da conexão que armazena as credenciais de autenticação para a fonte.
`objects`	Obrigatório. Uma lista de tabelas para ingestão. Cada entrada tem o formato `{"table": {"source_table": "..."}}`. Você também pode especificar um `destination_table` opcional dentro do objeto `table`.
`destination_catalog`	O catálogo onde as tabelas ingeridas são gravadas. Adota como default o catálogo definido durante a criação do pipeline.
`destination_schema`	O esquema em que as tabelas ingeridas são gravadas. Adota como default o esquema definido durante a criação do pipeline.
`scd_type`	A estratégia de dimensões que mudam lentamente (SCD): `SCD_TYPE_1`, `SCD_TYPE_2` ou `APPEND_ONLY`. O default é `SCD_TYPE_1`.
`primary_keys`	Substituir as chaves primárias default de uma tabela. Forneça uma lista de nomes de colunas.

Contribua com seu conector para a comunidade

Contribuir com seu conector para a comunidade é opcional. Seu conector está em execução em seu Workspace sem ele. Se você deseja compartilhar seu conector para que outros usuários possam descobri-lo e usá-lo, abra um pull request no repository Lakeflow comunidade Connectors. Os conectores contribuídos tornam-se conectores da comunidade, que a comunidade mantém e que não são suportados pelos SLAs da Databricks.

Requisitos​

Configurar o repository​

Implementar a interface LakeflowConnect​

Descrições de métodos​

Desenvolva seu conector​

Testar seu conector​

Conjunto de testes genérico (obrigatório)​

Teste de write-back (recomendado)​

Testes de unidade​

Construir o artefato de implantação​

Criar um pipeline de ingestão​

Opções de configuração do pipeline​

Contribua com seu conector para a comunidade​