Tutorial: SDK nativo para conectores de modelo Java¶

Introdução¶

Bem-vindo ao nosso tutorial sobre como usar um modelo de conector utilizando Snowflake Native SDK for Connectors. Este guia ajudará você a configurar um Connector Native Application.

Neste tutorial você aprenderá a:

• Implantar um Connector Native Application

• Configurar um conector de modelo para ingerir dados

• Personalizar um conector de modelo de acordo com suas próprias necessidades

O modelo contém vários comentários úteis no código para facilitar a localização de arquivos específicos que precisam ser modificados. Procure os comentários com as seguintes palavras-chave, elas irão orientá-lo e ajudar a implementar seu próprio conector:

• TODO

• TODO: HINT

• TODO: IMPLEMENT ME

Antes de começar este tutorial, você deve se preparar revisando o seguinte conteúdo recomendado:

• Snowflake Native SDK para conectores

• Início rápido do Snowflake Labs para criar e implantar um conector usando Native Apps Framework no Snowflake e Native Connectors SDK Java.

Inicialização e implantação¶

Para inicializar um projeto, clone o repositório Native SDK for Connectors do GitHub e copie o diretório /templates/native-sdk-connectors-java-template para o local do projeto desejado. Este modelo contém todo o código necessário para implantar um Connector Native Application. Feito isso, o modelo estará pronto para ser implantado.

make reinstall_application_from_version_dir

Etapa de pré-requisitos¶

Logo após a implantação, o Connector está na fase de Assistente. Esta fase consiste em algumas etapas que orientam o usuário final por todas as configurações necessárias. O primeiro passo é a etapa de pré-requisitos. É opcional e pode não ser necessário para todos os conectores. Os pré-requisitos geralmente são ações exigidas do usuário fora do aplicativo, por exemplo, executar consultas na planilha, fazer algumas configurações no sistema de origem etc.

Saiba mais sobre os pré-requisitos:

• Pré-requisitos

O conteúdo de cada pré-requisito é recuperado diretamente da tabela interna (STATE.PREREQUISITES) dentro do conector. Eles podem ser personalizados através do script setup.sql. No entanto, tenha em mente que o script setup.sql é executado em cada instalação, atualização e downgrade do aplicativo. As inserções devem ser idempotentes, por isso é recomendado utilizar uma consulta de mescla como no exemplo abaixo:

MERGE INTO STATE.PREREQUISITES AS dest
USING (SELECT * FROM VALUES
           ('1',
            'Sample prerequisite',
            'Prerequisites can be used to notice the end user of the connector about external configurations. Read more in the SDK documentation below. This content can be modified inside `setup.sql` script',
            'https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/flow/prerequisites',
            NULL,
            NULL,
            1
           )
) AS src (id, title, description, documentation_url, learnmore_url, guide_url, position)
ON dest.id = src.id
WHEN NOT MATCHED THEN
    INSERT (id, title, description, documentation_url, learnmore_url, guide_url, position)
    VALUES (src.id, src.title, src.description, src.documentation_url, src.learnmore_url, src.guide_url, src.position);

Etapa de configuração do Connector¶

A próxima etapa da fase do Assistente é a etapa de configuração do conector. Durante esta etapa, você pode configurar objetos de banco de dados e permissões necessárias para o conector. Esta etapa permite que as seguintes propriedades de configuração sejam especificadas:

• warehouse

• destination_database

• destination_schema

• operational_warehouse

• global_schedule

• data_owner_role

• agent_username

• agent_role

Se você precisar de outras propriedades personalizadas, elas poderão ser configuradas em uma das próximas etapas da fase do Assistente. Para mais informações sobre cada uma das propriedades, consulte:

• Configuração do conector

Além disso, o componente streamlit (streamlit/wizard/connector_config.py) fornecido no modelo mostra como acionar permissions-sdk e solicita algumas concessões do usuário final. Desde que as propriedades disponíveis satisfaçam as necessidades do conector, não há necessidade de substituir nenhuma das classes de back-end, embora isso ainda seja possível da mesma forma que para os componentes nas etapas posteriores da configuração.

Para mais informações sobre procedimentos internos e objetos Java, consulte:

• Referência de configuração do conector

O exemplo streamlit fornecido permite solicitar concessões em nível de conta como create database e execute tasks. Ele também permite que o usuário especifique uma referência de warehouse por meio do popup permissions-sdk.

No modelo, o usuário é solicitado a fornecer apenas o destination_database e destination_schema. No entanto, um comentário TODO em streamlit/wizard/connector_configuration.py contém código comentado que pode ser reutilizado para exibir mais caixas de entrada na UI do streamlit.

# TODO: Here you can add additional fields in connector configuration. Supported values are the following: warehouse, operational_warehouse, data_owner_role, agent_role, agent_username
# For example:
st.subheader("Operational warehouse")
input_col, _ = st.columns([2, 1])
with input_col:
    st.text_input("", key="operational_warehouse", label_visibility="collapsed")
st.caption("Name of the operational warehouse to be used")

Etapa de configuração da conexão¶

A próxima etapa da fase do Assistente é a etapa de configuração da conexão. Esta etapa permite que o usuário final configure parâmetros de conectividade externa para o conector. Esta configuração pode incluir identificadores de objetos como segredos, integrações etc. Como isso varia dependendo do sistema de origem dos dados ingeridos pelo conector, este é o primeiro lugar onde personalizações maiores precisam ser feitas no código-fonte.

Para obter mais informações sobre a configuração da conexão, consulte:

• Configuração de conexão

• Referência de configurações de conexão

Começando com o lado da UI streamlite (arquivo streamlit/wizard/connection_config.py) você precisa adicionar caixas de texto para todos os parâmetros necessários. Uma caixa de texto de exemplo é implementada para você e se você pesquisar o código neste arquivo, poderá encontrar um TODO com código comentado para um novo campo.

# TODO: Additional configuration properties can be added to the UI like this:
st.subheader("Additional connection parameter")
input_col, _ = st.columns([2, 1])
with input_col:
    st.text_input("", key="additional_connection_property", label_visibility="collapsed")
st.caption("Some description of the additional property")

Depois que as propriedades são adicionadas ao formulário, elas precisam ser passadas para a camada de backend do conector. Para fazer isso, dois locais adicionais devem ser modificados nos arquivos streamlit. O primeiro é a função finish_config no arquivo streamlit/wizard/connection_config.py. O estado das caixas de texto recém-adicionadas deve ser lido aqui. Além disso, ele pode ser validado se necessário e, em seguida, passado para a função set_connection_configuration. Por exemplo se additional_connection_property fosse adicionado, ficaria assim após as edições:

def finish_config():
try:
    # TODO: If some additional properties were specified they need to be passed to the set_connection_configuration function.
    # The properties can also be validated, for example, check whether they are not blank strings etc.
    response = set_connection_configuration(
        custom_connection_property=st.session_state["custom_connection_property"],
        additional_connection_property=st.session_state["additional_connection_property"],
    )

# rest of the method without changes

Depois a função set_connection_configuration deve ser editada, ela pode ser encontrada no arquivo streamlit/native_sdk_api/connection_config.py. Esta função é um proxy entre a UI streamlit e o procedimento SQL subjacente, que é um ponto de entrada para o backend do conector.

def set_connection_configuration(custom_connection_property: str, additional_connection_property: str):
    # TODO: this part of the code sends the config to the backend so all custom properties need to be added here
    config = {
        "custom_connection_property": escape_identifier(custom_connection_property),
        "additional_connection_property": escape_identifier(additional_connection_property),
    }

    return call_procedure(
        "PUBLIC.SET_CONNECTION_CONFIGURATION",
        [variant_argument(config)]
    )

Depois de fazer isso, a nova propriedade é salva na tabela de conectores internos, que contém a configuração. No entanto, as personalizações possíveis não param por aí. Alguns componentes de backend também podem ser personalizados. Procure os seguintes comentários no código para encontrá-los:

• TODO: IMPLEMENT ME connection configuration validate

• TODO: IMPLEMENT ME connection callback

• TODO: IMPLEMENT ME test connection

A parte de validação permite qualquer validação adicional nos dados recebidos da UI. Ela também pode transformar os dados, por exemplo, tornando-os minúsculos, cortando ou verificando se os objetos com os nomes fornecidos realmente existem dentro do Snowflake.

O retorno de chamada de conexão é uma parte que permite executar qualquer operação adicional com base na configuração, por exemplo, alterar procedimentos que precisam usar integrações de acesso externo.

A conexão de teste é um componente final da configuração da conexão, ela verifica se a conexão pode ser estabelecida entre o conector e o sistema de origem.

Para mais informações sobre esses componentes internos, consulte:

• Configuração de conexão

• Referência de configurações de conexão

Exemplos de implementações podem ser assim:

public class TemplateConfigurationInputValidator implements ConnectionConfigurationInputValidator {

    private static final String ERROR_CODE = "INVALID_CONNECTION_CONFIGURATION";

    @Override
    public ConnectorResponse validate(Variant config) {
      // TODO: IMPLEMENT ME connection configuration validate: If the connection configuration input
      // requires some additional validation this is the place to implement this logic.
      // See more in docs:
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/reference/connection_configuration_reference
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/flow/connection_configuration
      var integrationCheck = checkParameter(config, INTEGRATION_PARAM, false);
      if (!integrationCheck.isOk()) {
        return integrationCheck;
      }

      var secretCheck = checkParameter(config, SECRET_PARAM, true);
      if (!secretCheck.isOk()) {
        return ConnectorResponse.error(ERROR_CODE);
      }

      return ConnectorResponse.success();
    }
}

public class TemplateConnectionConfigurationCallback implements ConnectionConfigurationCallback {

    private final Session session;

    public TemplateConnectionConfigurationCallback(Session session) {
      this.session = session;
    }

    @Override
    public ConnectorResponse execute(Variant config) {
      // TODO: If you need to alter some procedures with external access you can use
      // configureProcedure method or implement a similar method on your own.
      // TODO: IMPLEMENT ME connection callback: Implement the custom logic of changes in application
      // to be done after connection configuration, like altering procedures with external access.
      // See more in docs:
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/reference/connection_configuration_reference
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/flow/connection_configuration
      configureProcedure(format("PUBLIC.TEST_CONNECTION()"), config);

      return ConnectorResponse.success();
    }
}

public class TemplateConnectionValidator {

    private static final String ERROR_CODE = "TEST_CONNECTION_FAILED";

    public static Variant testConnection(Session session) {
      // TODO: IMPLEMENT ME test connection: Implement the custom logic of testing the connection to
      // the source system here. This usually requires connection to some webservice or other external
      // system. It is suggested to perform only the basic connectivity validation here.
      // If that's the case then this procedure must be altered in TemplateConnectionConfigurationCallback first.
      // See more in docs:
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/reference/connection_configuration_reference
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/flow/connection_configuration
      return test().toVariant();
    }

    private static ConnectorResponse test() {
      try {
        var response = SourceSystemHttpHelper.testEndpoint();

        if (isSuccessful(response.statusCode())) {
          return ConnectorResponse.success();
        } else {
          return ConnectorResponse.error(ERROR_CODE, "Connection to source system failed");
        }
      } catch (Exception exception) {
        return ConnectorResponse.error(ERROR_CODE, "Test connection failed");
      }
    }
}

Etapa de finalização de configuração¶

A etapa de finalização da configuração do conector é a etapa final da fase do Assistente. Esta etapa tem múltiplas responsabilidades. Primeiro, ela permite que o usuário especifique qualquer configuração adicional necessária para o conector. Em segundo lugar, ela cria um banco de dados coletor, um esquema e, se necessário, algumas tabelas e exibições para os dados ingeridos. Por fim, ela inicializa componentes internos, como o planejador e o reator de tarefas.

Para mais informações sobre a finalização da configuração, consulte:

• Finalização da configuração

• Finalização da referência de configuração

Para mais informações sobre o reator de tarefas e agendamento, consulte:

• Reator de tarefa

• Referência SQL do reator de tarefas

• Agendador de ingestão

• Referência do agendador de ingestão

Semelhante à etapa de configuração da conexão, a personalização pode ser iniciada com a UI streamlit. streamlit/wizard/finalize_config.py contém um formulário com uma propriedade de exemplo. Mais propriedades podem ser adicionadas de acordo com as necessidades do conector. Para adicionar outra propriedade, procure por um comentário TODO, que contém código de exemplo de adição de uma nova propriedade no arquivo mencionado.

# TODO: Here you can add additional fields in finalize connector configuration.
# For example:
st.subheader("Some additional property")
input_col, _ = st.columns([2, 1])
with input_col:
    st.text_input("", key="some_additional_property", label_visibility="collapsed")
st.caption("Description of some new additional property")

Depois de adicionar a caixa de texto para uma nova propriedade, ela precisa ser passada para o backend. Para fazer isso, modifique a função finalize_configuration no mesmo arquivo:

def finalize_configuration():
    try:
        st.session_state["show_main_error"] = False
        # TODO: If some additional properties were introduced, they need to be passed to the finalize_connector_configuration function.
        response = finalize_connector_configuration(
            st.session_state.get("custom_property"),
            st.session_state.get("some_additional_property")
        )

Em seguida, abra streamlit/native_sdk_api/finalize_config.py e adicione-o à seguinte função:

def finalize_connector_configuration(custom_property: str, some_additional_property: str):
    # TODO: If some custom properties were configured, then they need to be specified here and passed to the FINALIZE_CONNECTOR_CONFIGURATION procedure.
    config = {
        "custom_property": custom_property,
        "some_additional_property": some_additional_property,
    }
    return call_procedure(
        "PUBLIC.FINALIZE_CONNECTOR_CONFIGURATION",
        [variant_argument(config)]
    )

Novamente, de forma semelhante à etapa de configuração de conexão, esta etapa também permite a personalização de vários componentes de backend, que podem ser encontrados usando as seguintes frases no código:

• TODO: IMPLEMENT ME validate source

• TODO: IMPLEMENT ME finalize internal

A parte de validação da fonte é responsável por executar validações mais sofisticadas nos sistemas de origem. Se a conexão de teste anterior apenas verificou se uma conexão pode ser estabelecida, a validação da origem poderá verificar o acesso a dados específicos no sistema, por exemplo, extraindo um único registro de dados.

A finalização interna é um procedimento interno responsável por inicializar o reator e o agendador de tarefas, criando um banco de dados de recebimento e os objetos aninhados necessários. Ele também pode ser usado para salvar a configuração fornecida durante a etapa de finalização (essa configuração não é salva por padrão).

Mais informações sobre os componentes internos podem ser encontradas em:

• Finalização da configuração

• Finalização da referência de configuração

Além disso, a entrada pode ser validada usando a interface FinalizeConnectorInputValidator e fornecê-la ao manipulador de finalização (verifique o TemplateFinalizeConnectorConfigurationCustomHandler). Mais informações sobre o uso de construtores podem ser encontradas em:

• Customização de procedimentos armazenados e manipuladores

Um exemplo de implementação da fonte de validação pode ser parecido com este:

public class SourceSystemAccessValidator implements SourceValidator {

    @Override
    public ConnectorResponse validate(Variant variant) {
      // TODO: IMPLEMENT ME validate source: Implement the custom logic of validating the source
      // system. In some cases this can be the same validation that happened in
      // TemplateConnectionValidator.
      // However, it is suggested to perform more complex validations, like specific access rights to
      // some specific resources here.
      // See more in docs:
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/reference/finalize_configuration_reference
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/flow/finalize_configuration
      var finalizeProperties = Configuration.fromCustomConfig(variant);

      var httpResponse = SourceSystemHttpHelper.validateSource(finalizeProperties.get("custom_property"));
      return prepareConnectorResponse(httpResponse.statusCode());
    }

    private ConnectorResponse prepareConnectorResponse(int statusCode) {
      switch (statusCode) {
        case 200:
          return ConnectorResponse.success();
        case 401:
          return ConnectorResponse.error("Unauthorized error");
        case 404:
          return ConnectorResponse.error("Not found error");
        default:
          return ConnectorResponse.error("Unknown error");
      }
    }
}

Criação de recursos¶

Após a conclusão da fase do assistente, o conector estará pronto para começar a ingerir dados. Mas primeiro, os recursos devem ser implementados e configurados. Um recurso é uma abstração que descreve um conjunto específico de dados no sistema de origem, por exemplo, uma tabela, um ponto de extremidade, um arquivo etc.

Diferentes sistemas de origem podem precisar de diferentes informações sobre um recurso, por esse motivo, uma definição de recurso precisa ser personalizada de acordo com as necessidades específicas. Para isso, vá até o arquivo streamlit/daily_use/data_sync_page.py. Lá você pode encontrar um TODO sobre a adição de caixas de texto para parâmetros de recursos. Os parâmetros de recursos devem permitir a identificação e recuperação de dados do sistema de origem. Esses parâmetros podem ser extraídos durante a ingestão:

# TODO: specify all the properties needed to define a resource in the source system. A subset of those properties should allow for a identification of a single resource, be it a table, endpoint, repository or some other data storage abstraction
st.text_input(
    "Resource name",
    key="resource_name",
)
st.text_input(
    "Some resource parameter",
    key="some_resource_parameter"
)

Depois que todas as propriedades necessárias forem adicionadas ao formulário, elas poderão ser passadas para o backend. Primeiro, o estado dos campos de texto deve ser extraído e passado para o método do nível da API queue_resource em streamlit/daily_use/data_sync_page.py:

def queue_resource():
    # TODO: add additional properties here and pass them to create_resource function
    resource_name = st.session_state.get("resource_name")
    some_resource_parameter = st.session_state.get("some_resource_parameter)

    if not resource_name:
        st.error("Resource name cannot be empty")
        return

    result = create_resource(resource_name, some_resource_parameter)
    if result.is_ok():
        st.success("Resource created")
    else:
        st.error(result.get_message())

Depois a função create_resource do streamlit/native_sdk_api/resource_management.py precisa ser atualizada:

def create_resource(resource_name, some_resource_parameter):
    ingestion_config = [{
        "id": "ingestionConfig",
        "ingestionStrategy": "INCREMENTAL",
        # TODO: HINT: scheduleType and scheduleDefinition are currently not supported out of the box, due to globalSchedule being used. However, a custom implementation of the scheduler can use those fields. They need to be provided becuase they are mandatory in the resourceDefinition.
        "scheduleType": "INTERVAL",
        "scheduleDefinition": "60m"
    }]
    # TODO: HINT: resource_id should allow identification of a table, endpoint etc. in the source system. It should be unique.
    resource_id = {
        "resource_name": resource_name,
    }
    id = f"{resource_name}_{random_suffix()}"

    # TODO: if you specified some additional resource parameters then you need to put them inside resource metadata:
    resource_metadata = {
        "some_resource_parameter": some_resource_parameter
    }

    return call_procedure("PUBLIC.CREATE_RESOURCE",
                          [
                              varchar_argument(id),
                              variant_argument(resource_id),
                              variant_list_argument(ingestion_config),
                              varchar_argument(id),
                              "true",
                              variant_argument(resource_metadata)
                          ])

Ingestão¶

Para realizar a ingestão de dados, você precisa implementar uma classe que manipulará a conexão com o sistema de origem e recuperará os dados, com base na configuração do recurso. Os módulos Scheduler e Task Reactor cuidarão do acionamento e enfileiramento de tarefas de ingestão.

A lógica de ingestão é invocada a partir da classe TemplateIngestion, procure por TODO: IMPLEMENT ME ingestion no código e substitua a geração de dados aleatórios pela recuperação de dados do sistema de origem. Se você adicionou algumas propriedades personalizadas à definição do recurso, elas podem ser obtidas das tabelas de conectores internos usando ResourceIngestionDefinitionRepository e propriedades disponíveis no TemplateWorkItem:

• resourceIngestionDefinitionId

• ingestionConfigurationId

Por exemplo, recuperar dados de algum webservice MIGHT parece com isso:

public final class SourceSystemHttpHelper {

  private static final String DATA_URL = "https://source_system.com/data/%s";
  private static final SourceSystemHttpClient sourceSystemClient = new SourceSystemHttpClient();
  private static final ObjectMapper objectMapper = new ObjectMapper();

  private static List<Variant> fetchData(String resourceId) {
    var response = sourceSystemClient.get(String.format(url, resourceId));
    var body = response.body();

    try {
        return Arrays.stream(objectMapper.readValue(body, Map[].class))
              .map(Variant::new)
              .collect(Collectors.toList());
    } catch (JsonProcessingException e) {
      throw new RuntimeException("Cannot parse json", e);
    }
  }
}

public class SourceSystemHttpClient {

  private static final Duration REQUEST_TIMEOUT = Duration.ofSeconds(15);

  private final HttpClient client;
  private final String secret;

  public SourceSystemHttpClient() {
    this.client = HttpClient.newHttpClient();
    this.secret =
        SnowflakeSecrets.newInstance()
            .getGenericSecretString(ConnectionConfiguration.TOKEN_NAME);
  }

  public HttpResponse<String> get(String url) {
    var request =
        HttpRequest.newBuilder()
            .uri(URI.create(url))
            .GET()
            .header("Authorization", format("Bearer %s", secret))
            .header("Content-Type", "application/json")
            .timeout(REQUEST_TIMEOUT)
            .build();

    try {
      return client.send(request, HttpResponse.BodyHandlers.ofString());
    } catch (IOException | InterruptedException ex) {
      throw new RuntimeException(format("HttpRequest failed: %s", ex.getMessage()), ex);
    }
  }
}

Gerenciamento do ciclo de vida dos recursos¶

Uma vez implementada a lógica de criação de recursos e sua ingestão, você pode gerenciar seu ciclo de vida pelos seguintes procedimentos:

1. PUBLIC.ENABLE_RESOURCE() – este procedimento habilita um recurso específico, o que significa que ele será agendado para ingestão

2. PUBLIC.DISABLE_RESOURCE() – este procedimento desabilita um recurso específico, o que significa que seu agendamento de ingestão será interrompido

3. PUBLIC.UPDATE_RESOURCE() – este procedimento permite atualizar as configurações de ingestão de um recurso específico. Ele não é implementado na UI do Streamlit por padrão porque às vezes pode ser indesejado pelo desenvolvedor permitir que o usuário do conector personalize a configuração de ingestão (revogue as concessões neste procedimento para a função de aplicativo ACCOUNTADMIN para proibir seu uso completamente).

Todos esses procedimentos têm manipuladores Java e são estendidos com retornos de chamada que permitem personalizar sua execução. É possível injetar implementações personalizadas de retornos de chamada usando o construtor desses manipuladores. Por padrão, o modelo fornece implementações Java simuladas de manipuladores de retorno de chamada para se livrar de procedimentos SQL de chamada que estendem todo o tempo de execução dos procedimentos mencionados. Essas implementações simuladas não fazem nada além de retornar uma resposta de sucesso. É possível fornecer a implementação personalizada para as classes de retorno de chamada preparadas pelo modelo ou criar esses retornos de chamada do zero e injetá-los no fluxo de execução do procedimento principal nos construtores de manipulador.

Configurações¶

O modelo contém uma aba de configurações que permite visualizar todas as configurações feitas anteriormente. No entanto, se as propriedades de configuração foram personalizadas, essa exibição também precisa de algumas personalizações. O código da aba Configurações pode ser encontrado no arquivo streamlit/daily_use/settings_page.py. Para personalizá-lo, basta extrair os valores da configuração das chaves que foram adicionadas nas respectivas configurações.

Por exemplo, se antes additional_connection_property foi adicionado na etapa de configuração da conexão, então poderia ser adicionado assim:

def connection_config_page():
    current_config = get_connection_configuration()

    # TODO: implement the display for all the custom properties defined in the connection configuration step
    custom_property = current_config.get("custom_connection_property", "")
    additional_connection_property = current_config.get("additional_connection_property", "")


    st.header("Connector configuration")
    st.caption("Here you can see the connector connection configuration saved during the connection configuration step "
               "of the Wizard. If some new property was introduced it has to be added here to display.")
    st.divider()

    st.text_input(
        "Custom connection property:",
        value=custom_property,
        disabled=True
    )
    st.text_input(
        "Additional connection property:",
        value=additional_connection_property,
        disabled=True
    )
    st.divider()