diff --git a/infra/elk/logstash_jdbc_input/Dockerfile b/infra/elk/logstash_jdbc_input/Dockerfile index 012226f..5ce0a17 100644 --- a/infra/elk/logstash_jdbc_input/Dockerfile +++ b/infra/elk/logstash_jdbc_input/Dockerfile @@ -8,5 +8,5 @@ RUN /usr/share/logstash/bin/logstash-plugin install logstash-filter-aggregate && RUN curl https://jdbc.postgresql.org/download/postgresql-42.2.18.jar -o $LOGSTASH_JDBC_DRIVER_JAR_LOCATION -ADD ./logstash.conf /usr/share/logstash/pipeline/logstash.conf +COPY ./*.conf /usr/share/logstash/pipeline/ diff --git a/infra/elk/logstash_jdbc_input/fts_authorities_ddjj.conf b/infra/elk/logstash_jdbc_input/fts_authorities_ddjj.conf new file mode 100644 index 0000000..cde22f2 --- /dev/null +++ b/infra/elk/logstash_jdbc_input/fts_authorities_ddjj.conf @@ -0,0 +1,33 @@ +input { + jdbc { + # AQE + jdbc_driver_library => "${LOGSTASH_JDBC_DRIVER_JAR_LOCATION}" + jdbc_driver_class => "${LOGSTASH_JDBC_DRIVER}" + jdbc_connection_string => "${LOGSTASH_JDBC_URL}" + jdbc_user => "${LOGSTASH_JDBC_USERNAME}" + jdbc_password => "${LOGSTASH_JDBC_PASSWORD}" + schedule => "* * * * *" + use_column_value => true + tracking_column => "id" + statement => 'SELECT id, document, last_name, first_name, year_elected, departament, charge, list, title, sex, nacionality, age, CAST(start as TEXT), CAST("end" as TEXT), presented, photo FROM analysis.authorities_with_ddjj WHERE id > :sql_last_value ORDER BY id' + last_run_metadata_path => "/logstash_data/ddjj.yml" + } +} +filter { + json { + source => "start" + target => "start" + } + json { + source => "end" + target => "end" + } +} +output { + elasticsearch { + hosts => ["${LOGSTASH_ELASTICSEARCH_HOST}"] + index => "fts_authorities_ddjj" + document_id => "%{id}" + } + stdout { codec => json_lines } +} diff --git a/infra/elk/logstash_jdbc_input/fts_full_data.conf b/infra/elk/logstash_jdbc_input/fts_full_data.conf new file mode 100644 index 0000000..2dec294 --- /dev/null +++ b/infra/elk/logstash_jdbc_input/fts_full_data.conf @@ -0,0 +1,24 @@ +input { + jdbc { + # full_data + jdbc_driver_library => "${LOGSTASH_JDBC_DRIVER_JAR_LOCATION}" + jdbc_driver_class => "${LOGSTASH_JDBC_DRIVER}" + jdbc_connection_string => "${LOGSTASH_JDBC_URL}" + jdbc_user => "${LOGSTASH_JDBC_USERNAME}" + jdbc_password => "${LOGSTASH_JDBC_PASSWORD}" + schedule => "* * * * *" + use_column_value => true + tracking_column => "document" + statement => "select * from analysis.full_data WHERE document > :sql_last_value ORDER BY document" + last_run_metadata_path => "/logstash_data/full_data-jdbc-int-sql_last_value.yml" + } +} + +output { + elasticsearch { + hosts => ["${LOGSTASH_ELASTICSEARCH_HOST}"] + index => "fts_full_data" + document_id => "%{document}" + } + stdout { codec => json_lines } +} diff --git a/infra/elk/logstash_jdbc_input/logstash.conf b/infra/elk/logstash_jdbc_input/logstash.conf deleted file mode 100644 index 3d66976..0000000 --- a/infra/elk/logstash_jdbc_input/logstash.conf +++ /dev/null @@ -1,64 +0,0 @@ -input { - jdbc { - # AQE - jdbc_driver_library => "${LOGSTASH_JDBC_DRIVER_JAR_LOCATION}" - jdbc_driver_class => "${LOGSTASH_JDBC_DRIVER}" - jdbc_connection_string => "${LOGSTASH_JDBC_URL}" - jdbc_user => "${LOGSTASH_JDBC_USERNAME}" - jdbc_password => "${LOGSTASH_JDBC_PASSWORD}" - schedule => "* 0 * * *" - statement => "SELECT regexp_replace(identifier, '\.', '', 'g') as document, name || ' ' || lastname as name, 'a_quien_elegimos' as source, head_shot as photo FROM staging.a_quien_elegimos" - last_run_metadata_path => "/logstash_data/aqe-jdbc-int-sql_last_value.yml" - } - - jdbc { - # Declarations - jdbc_driver_library => "${LOGSTASH_JDBC_DRIVER_JAR_LOCATION}" - jdbc_driver_class => "${LOGSTASH_JDBC_DRIVER}" - jdbc_connection_string => "${LOGSTASH_JDBC_URL}" - jdbc_user => "${LOGSTASH_JDBC_USERNAME}" - jdbc_password => "${LOGSTASH_JDBC_PASSWORD}" - schedule => "* * * * *" - use_column_value => true - tracking_column => "id" - statement => "SELECT id, 'declarations' as source, document, name, net_worth, active, passive FROM analysis.declarations WHERE id > :sql_last_value" - last_run_metadata_path => "/logstash_data/declarations-jdbc-int-sql_last_value.yml" - } - - jdbc { - # ANDE - jdbc_driver_library => "${LOGSTASH_JDBC_DRIVER_JAR_LOCATION}" - jdbc_driver_class => "${LOGSTASH_JDBC_DRIVER}" - jdbc_connection_string => "${LOGSTASH_JDBC_URL}" - jdbc_user => "${LOGSTASH_JDBC_USERNAME}" - jdbc_password => "${LOGSTASH_JDBC_PASSWORD}" - schedule => "* * * * *" - use_column_value => true - tracking_column => "id" - statement => "SELECT id, documento as document, cliente as name, 'ande_exonerados' as source FROM staging.ande_exonerados WHERE id > :sql_last_value" - last_run_metadata_path => "/logstash_data/ande_exonerados-jdbc-int-sql_last_value.yml" - } - - jdbc { - # TSJE - jdbc_driver_library => "${LOGSTASH_JDBC_DRIVER_JAR_LOCATION}" - jdbc_driver_class => "${LOGSTASH_JDBC_DRIVER}" - jdbc_connection_string => "${LOGSTASH_JDBC_URL}" - jdbc_user => "${LOGSTASH_JDBC_USERNAME}" - jdbc_password => "${LOGSTASH_JDBC_PASSWORD}" - schedule => "* * * * *" - use_column_value => true - tracking_column => "id" - statement => "SELECT id, nombre || ' ' || apellido as name, cedula as document, edad as age, 'tsje_elected' as source FROM analysis.tsje_elected WHERE id > :sql_last_value" - last_run_metadata_path => "/logstash_data/tsje_elected-jdbc-int-sql_last_value.yml" - } -} - -output { - elasticsearch { - hosts => ["${LOGSTASH_ELASTICSEARCH_HOST}"] - index => "fts_people" - document_id => "%{id}" - } - stdout { codec => json_lines } -} diff --git a/infra/elk/re-index.sh b/infra/elk/re-index.sh new file mode 100644 index 0000000..c204673 --- /dev/null +++ b/infra/elk/re-index.sh @@ -0,0 +1,24 @@ +#!/bin/bash + + +set -e +set -x + +INDEX_NAME=$1 +FILE_NAME=$2 + + +docker-compose stop logstash + +rm -rvf "logstash_data/$FILE_NAME" + +docker-compose exec -T elasticsearch curl -XDELETE localhost:9200/$INDEX_NAME + +echo "Index removed, restarting" + + +docker-compose up -d --build logstash + +echo "The index should start updating very soon, check logs" + + diff --git a/scripts/python/airflow/dags/elastic_fts_auth_ddjj.py b/scripts/python/airflow/dags/elastic_fts_auth_ddjj.py new file mode 100644 index 0000000..f38bdf7 --- /dev/null +++ b/scripts/python/airflow/dags/elastic_fts_auth_ddjj.py @@ -0,0 +1,42 @@ +from datetime import timedelta, datetime + +from airflow import DAG +from airflow.operators.bash_operator import BashOperator +from airflow.operators.postgres_operator import PostgresOperator + +default_args = { + 'owner': 'airflow', + 'depends_on_past': False, + 'email': ['arturovolpe@gmail.com'], + 'email_on_failure': False, + 'email_on_retry': False, + 'retries': 1, + 'retry_delay': timedelta(seconds=5), + 'params': { + } +} +dag = DAG( + 'elastic_fts_auth_ddjj', + default_args=default_args, + description='ETL that creates the table for search authorities with ddjj and then indexes the table in a elastic', + start_date=datetime(2020, 12, 21), + schedule_interval=timedelta(weeks=1), +) + +with dag: + do_curl = BashOperator( + task_id=f'call_webhook', + bash_command=""" + curl "{{ var.value.ELASTIC_IDX_AUTH_DDJJ }}" + """, + retries=10 + ) + + do_query = PostgresOperator(task_id='do_query', + sql="sql/elastic_index_fts_auth_ddjj.sql") + + do_query >> do_curl + +if __name__ == '__main__': + dag.clear(reset_dag_runs=True) + dag.run() diff --git a/scripts/python/airflow/dags/elastic_fts_full_data_index.py b/scripts/python/airflow/dags/elastic_fts_full_data_index.py new file mode 100644 index 0000000..c93da3d --- /dev/null +++ b/scripts/python/airflow/dags/elastic_fts_full_data_index.py @@ -0,0 +1,46 @@ +from datetime import timedelta, datetime + +from airflow import DAG +from airflow.operators.bash_operator import BashOperator +from airflow.operators.postgres_operator import PostgresOperator + +default_args = { + 'owner': 'airflow', + 'depends_on_past': False, + 'email': ['arturovolpe@gmail.com'], + 'email_on_failure': False, + 'email_on_retry': False, + 'retries': 1, + 'retry_delay': timedelta(seconds=5), + 'params': { + } +} +dag = DAG( + 'elastic_fts_full_data_index', + default_args=default_args, + description='ETL that creates the table with all people data', + start_date=datetime(2020, 12, 21), + schedule_interval=timedelta(weeks=1), +) + +with dag: + do_curl = BashOperator( + task_id=f'call_webhook', + bash_command=""" + curl "{{ var.value.ELASTIC_IDX_FULL_DATA_HOOK }}" + """, + retries=10 + ) + + clean_db = PostgresOperator(task_id='clean_table', + sql="DROP TABLE analysis.full_data") + + do_query = PostgresOperator(task_id='do_query', + + sql="sql/elastic_index_full_data.sql") + + clean_db >> do_query >> do_curl + +if __name__ == '__main__': + dag.clear(reset_dag_runs=True) + dag.run() diff --git a/scripts/python/airflow/dags/sql/elastic_index_fts_auth_ddjj.sql b/scripts/python/airflow/dags/sql/elastic_index_fts_auth_ddjj.sql new file mode 100644 index 0000000..0a9cb04 --- /dev/null +++ b/scripts/python/airflow/dags/sql/elastic_index_fts_auth_ddjj.sql @@ -0,0 +1,52 @@ +DROP TABLE analysis.authorities_with_ddjj; + +CREATE TABLE analysis.authorities_with_ddjj AS ( + SELECT autoridad.id, + autoridad.cedula AS document, + autoridad.apellido AS last_name, + autoridad.nombre AS first_name, + autoridad.ano AS year_elected, + autoridad.dep_desc AS departament, + autoridad.cand_desc AS charge, + autoridad.nombre_lista AS list, + autoridad.desc_tit_sup AS title, + autoridad.sexo AS sex, + autoridad.nacionalidad AS nacionality, + autoridad.edad AS age, + (SELECT jsonb_build_object( + 'id', start.id, + 'link', start.link, + 'link_sandwich', start.link_sandwich, + 'origin', start.origin, + 'active', start.active, + 'passive', start.passive, + 'net_worth', start.net_worth) + FROM analysis.declarations start + WHERE start.document = autoridad.cedula + AND start.year = autoridad.ano + ORDER BY start.version desc + LIMIT 1) AS start, + (SELECT jsonb_build_object( + 'id', "end".id, + 'link', "end".link, + 'link_sandwich', "end".link_sandwich, + 'origin', "end".origin, + 'active', "end".active, + 'passive', "end".passive, + 'net_worth', "end".net_worth) + FROM analysis.declarations "end" + WHERE "end".document = autoridad.cedula + AND "end".year = autoridad.ano + 5 + ORDER BY "end".version desc + LIMIT 1) AS "end", + aqe.head_shot as photo + FROM analysis.tsje_elected autoridad + LEFT JOIN staging.a_quien_elegimos aqe ON aqe.identifier = autoridad.cedula +); + +ALTER TABLE analysis.authorities_with_ddjj + ADD COLUMN presented boolean DEFAULT FALSE; + +UPDATE analysis.authorities_with_ddjj +SET presented = start IS NOT NULL AND ("end" IS NOT NULL OR year + 5 > date_part('year', CURRENT_DATE)); +; diff --git a/migrations/R_Full_data_table.sql b/scripts/python/airflow/dags/sql/elastic_index_full_data.sql similarity index 98% rename from migrations/R_Full_data_table.sql rename to scripts/python/airflow/dags/sql/elastic_index_full_data.sql index 4f585fa..e9efb87 100644 --- a/migrations/R_Full_data_table.sql +++ b/scripts/python/airflow/dags/sql/elastic_index_full_data.sql @@ -1,4 +1,3 @@ -DROP TABLE analysis.full_data; CREATE TABLE analysis.full_data AS ( WITH sfp_documents AS ( select documento as document, @@ -123,7 +122,7 @@ CREATE TABLE analysis.full_data AS ( (raw.ano || to_char(raw.mes, '00')) = docs.last_month_worked GROUP BY docs.document, docs.last_month_worked ) - SELECT CAST(regexp_replace(document, '[^0-9]+', '', 'g') as bigint) as document, + SELECT regexp_replace(document, '[^0-9]+', '', 'g') as document, array_agg(name) as name, array_agg(photo) as photo, array_agg(salary) as salary,