feat: Add ETL to re-index the elasticsearch indices

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/
 

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 }
+}

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 }
+}

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"
+
+

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

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

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));
+;

migrations/R_Full_data_table.sql → 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,