python ETL tools pyetl

from pyetl import Task, DatabaseReader, DatabaseWriter
reader = DatabaseReader("sqlite:///db1.sqlite3", table_name="source")
writer = DatabaseWriter("sqlite:///db2.sqlite3", table_name="target")
Task(reader, writer).start()

from pyetl import Task, DatabaseReader, HiveWriter2
reader = DatabaseReader("sqlite:///db1.sqlite3", table_name="source")
writer = HiveWriter2("hive://localhost:10000/default", table_name="target")
Task(reader, writer).start()

from pyetl import Task, DatabaseReader, ElasticSearchWriter
reader = DatabaseReader("sqlite:///db1.sqlite3", table_name="source")
writer = ElasticSearchWriter(hosts=["localhost"], index_name="tartget")
Task(reader, writer).start()

# Original table source contains uuid, full_name fields
reader = DatabaseReader("sqlite:///db.sqlite3", table_name="source")
# The target table target contains the id, name fields
writer = DatabaseWriter("sqlite:///db.sqlite3", table_name="target")
# columns configure the field mapping relationship between the target table and the original table
columns = {"id": "uuid", "name": "full_name"}
Task(reader, writer, columns=columns).start()

# functionsConfigure the udf mapping of the fields as follows: id to string, name with spaces removed
functions={"id": str, "name": lambda x: ()}
Task(reader, writer, columns=columns, functions=functions).start()

import json
from pyetl import Task, DatabaseReader, DatabaseWriter

class NewTask(Task):
  reader = DatabaseReader("sqlite:///db.sqlite3", table_name="source")
  writer = DatabaseWriter("sqlite:///db.sqlite3", table_name="target")
  
  def get_columns(self):
    """Generate field mapping configurations by means of functions for more flexible use"""
    # The following example takes the field mapping configuration out of the database and returns it as a dictionary type
    sql = "select columns from task where name='new_task'"
    columns = .read_one(sql)["columns"]
    return (columns)
   
  def get_functions(self):
    """Generating udf mappings for fields by means of functions"""
    # The following example converts each field type to a string
    return {col: str for col in }
   
  def apply_function(self, record):
    """udf of an entire piece of data in a data stream"""
    record["flag"] = int(record["id"]) % 2
    return record

  def before(self):
    """Operations to be performed before the task starts, such as initializing the task table, creating the target table, etc."""
    sql = "create table destination_table(id int, name varchar(100))"
    (sql)
  
  def after(self):
    """Actions to be performed after the task is completed, such as updating the task status, etc."""
    sql = "update task set status='done' where name='new_task'"
    (sql)

NewTask().start()