aphp / spark-etl Goto Github PK

being able to run from pySpark

Right now the csv is only written locally.
It should go into hdfs by default and locally if prefixed with "file://"

There might be good ideas in the sqoop direct import in particular the PostgresqlAsyncSink class

This will add a new column that calculate the hash of the columns.

A function would have :

• dataframe
• hashColumnName
• column not to hash[]

Might be interesting to:

• load dataframe from csv with arrays
• input/output from table with array columns

some ideas there:
read
write

• temp folder
• temp csv
• spark temp tables
• postgres temp tables

By splitting a table in multiple query from multiple executors
this will speedup the data retrieval. Steps are:

1. get min/max
2. setup multiple executor
3. run multiple COPY FROM

when we try to use this tool following instructions from readme, I am getting
java.lang.ClassNotFoundException: org.apache.spark.internal.Logging$class Is there any way to fix this ?

postgres returns f/t for boolean
transform it to false /true

Allow deindexing before large bulk load and reindex the table afterward

• Deindex

UPDATE pg_index 
SET indisready=false 
WHERE indrelid IN (
SELECT pg_class.oid FROM pg_class
JOIN pg_catalog.pg_namespace n ON n.oid = pg_class.relnamespace
WHERE relname='theTable' and nspname = 'theSchema'
);

• Insert
• Reindex

UPDATE pg_index 
SET indisready=true 
WHERE indrelid IN (
SELECT pg_class.oid FROM pg_class
JOIN pg_catalog.pg_namespace n ON n.oid = pg_class.relnamespace
WHERE relname='theTable' and nspname = 'theSchema'
);

REINDEX TABLE theSchema.theTable;

Parallel level would be the number of concurrent postgres connection
while
partitions would be the resulting number of csv in hdfs.

This will be useful in case of large multiline csv, because they will be splitted into multiple
small csv instead of few large ones.

goal is to let spark distinguish the changes, insert new rows, and update different rows. Steps are:

1. fetch both ID and HASH from the entire target table to spark
2. compare ID and HASH from target table and source table:
   1. create new rows
   2. create rows to update
3. insert new rows directly
4. insert rows to update in a temp table, and update

This way limits the postgresql computation.

being able to run from sparkR