gestaogovbr / fastetl Goto Github PK

The GeoPointDataCleaner fails with an error if the data or QA dataframe are empty. It should instead end let the Airflow task succeed.

When the QA dataframe in the GeoPointDataCleaner class is empty, it propagates an error because it will generate a data patch file with nans for primary keys.

The empty QA dataframe cases must be handled correctly and do not generate a data patch file.

The function _build_incremental_sqls is named as an internal function (starts with _) but is actually imported directly and used by external modules.

It should be renamed appropriately.

What is this module? How can I install it? How do I use it with Apache Airflow?

We need to create a README file with some basic information about the module.

Since PR #104 some continuous integration tests have been failing. It seems like the failing tests are the ones that test full table replication in CopyDbToDbOperator.

The usage of next_execution_date in a context has been deprecated in Airflow. Some warnings tell us to replace it with data_interval_end.

The templates defined in custom_functions.utils.date.py break in Airflow 2.2 when running regularly scheduled dags (without a manual trigger).

O FastETL já tem o LoadGSheetCSVOperator para copiar uma planilha Google para um arquivo CSV.

Seria desejável ter um operador também para fazer o processo inverso.

Precisamos decidir um formato padrão para usar nas docstrings do repositório, documentá-lo nas instruções de contribuição e adaptar as docstrings em todos os módulos Python para que fiquem em conformidade com essa decisão.

Algumas possibilidades de estilo de docstring:

• Epytext
• reST (exemplo)
• Google (exemplo)
• numpy (exemplo)

A PEP 287 recomenda o estilo reST. Porém muitos módulos deste repositório já estão usando Google style docstrings.

Mais informações:

• What is the standard Python docstring format (Stack Overflow)
• Python docstrings styles - Krystian Safjan's Blog

Atualmente o parâmetro chunksize faz com que a leitura dos dados (query SELECT) seja feita em lote. Mas a gravação (query UPDATE) ainda está sendo feita linha a linha. É provável que fazê-lo em lotes implique em ganhos de desempenho.

Using execution_date inside templates is deprecated in Airflow 2.2.1 and will be removed in a future version. We need to replace it with an alternative.

Currently it throws a warning:

{warnings.py:109} WARNING - /home/***/.local/lib/python3.9/site-packages/***/models/taskinstance.py:1941: DeprecationWarning: Accessing 'execution_date' from the template is deprecated and will be removed in a future version. Please use 'logical_date' or 'data_interval_start' instead.
  warnings.warn(message, DeprecationWarning)

A recently change in FastETL caused it to no longer be compatible with Airflow 1.x. We need to find out the minimum required version of Airflow and write it in the documentation.

Atualmente o OSRMDistanceDbOperator sempre escreve a coluna de destino, não importando se nela já constam valores.

Seria interessante ter um parâmetro booleano chamado overwrite_existing para escolher se os valores existentes na coluna serão sobrescritos. Caso seja False, somente valores NULL serão alterados na coluna.

In order to add CI/CD to this library, we should create tests that verify that the hooks and operators are working as expected.

The Astronomer blog has a post with instructions and tips for adding unit tests to Airflow hooks and operators.

If OrgaosMappingCleaner receives an empty dataframe (e.g. when doing an incremental database load and there is no change in data), it fails with the error

KeyError: "None of [Index(['nan'], dtype='object', name='key_name')] are in the [index]"

It should instead handle gracefully, log the occurrence of an empty dataframe and return without error so that the Airflow task can succeed.

Os testes do CopyDbToDbOperator abrangem somente Postgres SQL e Microsoft SQL Server. É necessário ter testes também para o MySQL.

Prover compatibilidade para os métodos copy_db_to_db e sync_db_to_db para conexões oriundas do MySQL. Atualmente os métodos suportam apenas Postgres e SQL Server.

Para contabilizar a quantidade de linhas processadas durante a carga (cópia) dos dados em modo full ou incremental, adicionar 4 parâmetros opcionais nas functions copy_db_to_db e sync_db_2_db:

• nome da conexão Airflow origem
• nome da conexão Airflow destino
• nome do schema da tabela de contabilização
• nome da tabela de contabilização

Para simplificar, a tabela de contabilização pode residir na mesma conexão da tabela destino.
Para o cálculo, , no caso de full, basta executar um count(*) na tabela ao final da carga. No caso de incremental, basta executar o count(*) na tabela intermediária que contém as linhas modificadas desde a carga anterior.

Ao usar o OSRMDistanceDbOperator com o banco de dados Microsoft SQL Server, a primeira query UPDATE fica travada esperando algo em deadlock, por causa da query SELECT que vem antes.

Segundo o @lucasbenevides, se usarmos WITH (NOLOCK) na query SELECT podemos evitar o problema.

• Automate the creation of the twin table and removal at the end
• Turn into an Airflow Operator
• Allow reading from MsSql and writing to Postgres
• Allow inserting load date into the destination table
• Create tests

The incremental load functions (_build_incremental_sqls and sync_db_2_db) currently only support a single primary key column. The parameter should be a list of primary keys and the functions should support multiple primary keys.

Create new hook in FastETL for updating open data resources with the new API of the Brazilian Open Data Portal (https://dados.gov.br/swagger-ui.html)

The merge_patches function of patchwork is not removing lines determined to be rejected in the QA file. The nivel_erro column is improperly being cast to str, when it's actually of type int64.

In file bacen_sta_hook, the content of the email after a updated password contains fixed personal login information. Needs to change it to get dynamically from Airflow Connections;

• Change Postgres connection to pyodbc
• Allow inserting the load date in the destination table

In the patchtowrk module, if the quality control dataframe is empty and someone calls DataPatch.to_file, the method fails with a confusing message:

KeyError: "None of [Index(['nan'], dtype='object', name='key_name')] are in the [index]"

It should instead fail with a clear and useful error message.

The test container fails to setup (make setup) on Github Actions because Airflow is not ready when it runs airflow db reset.

A proposed solution is to build a new container inheriting from the Airflow container that add the airflow db init step before the setup phase.

A biblioteca oauth2client foi acrescentada no ambiente pois o GSheetHook a importa diretamente (até então estava sendo instalada pela versão antiga da pygsheets==2.0.3,1). Porém esta versão está depreciada e foi removida da pygsheets. Remover todos os usos que importam oauth2client e substituir pela alternativa atual.

Casos em que a rota não pôde ser calculada (ex.: coordenadas fora do espaço para o qual o servidor possui mapas) estão retornando distância zero. O correto seria retornar None para indicar que não foi possível calcular a rota.

O operador OSRMDistanceDbOperator precisa lidar com a possibilidade de alguns valores nas colunas das coordenadas de origem e destino serem NULL. Nesse caso, a distância retornada deve também ser NULL.

This project should specify version numbers so that others that depend on it can use dependency control.

The history of changes in each new version should be registered in a CHANGELOG.

O DownloadCSVFromDbOperator salva a tabela em formato csv, que não preserva os tipos das colunas. Por exemplo, se no banco uma coluna é do tipo VARCHAR, mas os dados são numéricos, ao salvar em csv e ler com o Pandas, este vai interpretar como Int64 ou float.

Precisamos de um outro Operator ou um parâmetro no DownloadCSVFromDbOperator que possibilite extrair uma tabela preservando os tipos das colunas.

Parameter column_list has wrong type hint: str, but is used as a list a few lines below.

When using the DbtoDpOperator to copy a table to a non-existing destination table, the create_table_if_not_exists command creates only the table but misses the columns, resulting in an error. This scenario occurred with the following table from Quartzo: cipiws_VBL.arquivo_intervencao. The table was created in the data lake but without any columns.

Adjust function create_table_if_not_exists for Teiid (Quartzo) sources. By now this function only works when the source connection provider of the database is PostgreSQL or SQL Server. This improvement will require a datatype mapping dictionary considering the destination sources can be both SQL Server or Postgres.

O empacotamento vai facilitar a instalação do FastETL por outras pessoas (através do pip) e garantir a resolução das dependências de outros pacotes.

Evoluir os plugins de replicação de tabelsa full e completa para automaticamente criarem (caso não exista) a tabela no banco de dados destino a partir da estrutura da tabela do banco de dados origem. Isso reduzirá a barreira para utilização dos plugins.

The DuplicatedRowCleaner fails with an error if there are no rows with duplicated keys. It should instead end let the Airflow task succeed.

Reference documentation is textual documentation on each available class and method offered by the library, preferably generated automatically from docstrings.

Decide upon a documentation system to use (e.g. Sphinx) and where to host it (e.g. readthedocs.org?).

Provide instructions on how to build the documentation.

For an example, see Apache Airflow's documentation.

Update function get_trigger_date to return the datetime with the server timezone configuration, and not UTC. This update is necessary to avoid problems when executing DAGs near the end of the day.

Env Var: AIRFLOW__CORE__DEFAULT_TIMEZONE

O parâmetro steps na API do OSRM determina se será incluída a descrição passo a passo da rota calculada.

Nos casos em que interessa apenas calcular a distância, poupa-se algum tráfego de rede ao não incluir esses passos na resposta json.

Quando tabelas com milhões de linhas recebem carga full, o processamento às vezes demora vários dias para completar. Sendo assim, a última data de referência disponível na tabela para o incremental é a do último dia da carga. Porém, neste caso, a data de referência correta deve ser a do primeiro dia, para que o incremental não descarte alterações realizadas desde o início do full.

Por isso, na function sync_db_2_db, é preciso adicionar um parâmetro opcional since_date que permita especificar esta data de referência. Quando informada, deverá se sobrepor ao resultado obtido na function _build_increm_filter que procura o max(<data-alteração>) da tabela.

In order to reach an international audience, many things need to be translated to English.

Let's begin with the most obvious: the README.md file. The original text in Portuguese should be moved to LEIAME.md;

Github issues a warning in Actions when we run actions/checkout@v2. It seems that that version is tied to node.js version 12 which is deprecated. Updating to v3 could solve the warning.

Incluir uma coluna Datetime na tabela de teste para testar as conversões com o Sql Server que dificulta a inserção de dados no tipo datetime2.

According to the Data Package specification, Package and Resource names must be lowercase.

When Patchwork creates a Data Package from database schemata and/or tables that contain uppercase characters, that will cause an error in Frictionless when writing the package (from Frictionless v5.x on).

Ao atualizar as linhas do banco de dados dentro de um loop de uma consulta SELECT, o banco SQL Server está entrando em deadlock e não passa da primeira consulta.

Ver Stackoverflow.

All Python modules used by this project should be documented in a requirements.txt file.

As per good practice in open repositories, this should have contributing guidelines in a CONTRIBUTING file, including:

• refer to the code of conduct (another document that is good practice, but we don't have it yet)
• instructions to fork the project and set up a developing environment
• instructions to follow the expected workflow of contributions (setting up a branch, pushing to your fork, creating a PR, etc.)
• remind contributors that by submitting a PR they are licensing their contributions with the repository's LICENSE

Similarmente ao CopyDbToDbOperator é necessário criar um outro operador para encapsular a lógica da função fast_etl.sync_db_2_db().