Have a problem with my ETL process. I've got ETL process, written in python and it works great, but operations starts one after another, so the whole process lasts much time. I'm slightly new in Apache Airflow, but I've made a DUG and there is a problem with him) I get a mistake:
File "/usr/lib/python3.8/encodings/utf_16_le.py", line 15, in decode
def decode(input, errors='strict'):
File "/usr/local/lib/python3.8/dist-packages/airflow/models/taskinstance.py", line 1543, in signal_handler
raise AirflowException("Task received SIGTERM signal")
airflow.exceptions.AirflowException: Task received SIGTERM signal
The above exception was the direct cause of the following exception:
airflow.exceptions.AirflowException: decoding with 'utf-16le' codec failed (AirflowException: Task received SIGTERM signal)
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/usr/local/lib/python3.8/dist-packages/sqlalchemy/engine/base.py", line 1705, in _execute_context
self.dialect.do_execute(
File "/usr/local/lib/python3.8/dist-packages/sqlalchemy/engine/default.py", line 716, in do_execute
cursor.execute(statement, parameters)
SystemError: <class 'pyodbc.Error'> returned a result with an error set
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/usr/local/lib/python3.8/dist-packages/sqlalchemy/engine/base.py", line 896, in _rollback_impl
self.engine.dialect.do_rollback(self.connection)
File "/usr/local/lib/python3.8/dist-packages/sqlalchemy/engine/default.py", line 666, in do_rollback
dbapi_connection.rollback()
pyodbc.OperationalError: ('08S01', '[08S01] [Microsoft][ODBC Driver 18 for SQL Server]Communication link failure (0) (SQLEndTran)')
Here is a code of my Task. There can be up to 10 connections at once:
def update_from_gladiator_ost(market_id):
query = "DELETE from [stage].[dbo].[rests_by_docs_temp] where market_id = %d" % market_id
execute_query_dwh(query)
engine = dwh_conn()
connection = engine.raw_connection()
abc = connection.cursor()
# abc.execute("DELETE from [stage].[dbo].[sell_movement_temp]; DELETE from [stage].[dbo].[rests_by_docs_temp]")
df_op = pd.read_sql(
"SET NOCOUNT ON exec [dbo].[mp_report_finance_agent_enhanced_basis_transport_royalty_NC_ost_by_docs4] @pmarket_id = %d, @pstart_date = '%s', @pend_date = '%s', @pselect = '1'" % (
market_id, z, w), gladiator_conn())
df_op = df_op.fillna(value=0)
for row_count in range(0, df_op.shape[0]):
chunk = df_op.iloc[row_count:row_count + 1, :].values.tolist()
tuple_of_tuples = tuple(tuple(x) for x in chunk)
abc.executemany(
"insert into stage.dbo.rests_by_docs_temp" + " ([date_start],[market_id],[good_id],[agent_id],[doc_id],[tstart_qty],[tstart_amt],[IMP],[doc_name]) values (?,?,?,?,?,?,?,?,?)",
tuple_of_tuples)
abc.commit()
connection.close()
As you see, I get data from database and INSERT it in my DWH
And here is my connections:
def dwh_conn():
mySQL = '192.168.240.1'
myDB = 'DWH'
login = 'sa'
PWD = '....'
Encrypt = 'No'
Certificate = 'Yes'
params = urllib.parse.quote_plus("DRIVER={ODBC Driver 18 for SQL Server};"
"SERVER=" + mySQL + ";"
"SERVER=" + mySQL + ";"
"Port=1433" + ";"
"DATABASE=" + myDB + ";"
"UID=" + login + ";"
"PWD=" + PWD + ";"
"Encrypt=" + Encrypt + ";"
"TrustServerCertificate=" + Certificate + ";")
engine = sa.create_engine('mssql+pyodbc:///?odbc_connect={}?charset=utf8mb4'.format(params), fast_executemany=True)
return engine
def gladiator_conn():
mySQL = '...'
myDB = '...'
login = '...'
PWD = '...'
Encrypt = 'No'
Certificate = 'Yes'
params = urllib.parse.quote_plus("DRIVER={ODBC Driver 18 for SQL Server};"
"SERVER=" + mySQL + ";"
"Port=1433" + ";"
"DATABASE=" + myDB + ";"
"UID=" + login + ";"
"PWD=" + PWD + ";"
"Encrypt=" + Encrypt + ";"
"TrustServerCertificate=" + Certificate + ";")
engine = sa.create_engine('mssql+pyodbc:///?odbc_connect={}?charset=utf8mb4'.format(params), fast_executemany=True)
return engine
I think the problem is in unixODBC. Because when I do the whole code in Pycharm on Windows - everythong is fine. But on docker Ubuntu/Airflow - it sometimes fails. I can restart the task which failed and it can go fine but can fail again
updated: I guess, I Found one solution but I cant realize it on my case.
def decode_sketchy_utf16(raw_bytes):
s = raw_bytes.decode("utf-16le", "ignore")
try:
n = s.index('\u0000')
s = s[:n] # respect null terminator
except ValueError:
pass
return s
# ...
prev_converter = cnxn.get_output_converter(pyodbc.SQL_WVARCHAR)
cnxn.add_output_converter(pyodbc.SQL_WVARCHAR, decode_sketchy_utf16)
col_info = crsr.columns("Clients").fetchall()
cnxn.add_output_converter(pyodbc.SQL_WVARCHAR, prev_converter) # restore previous behaviour
Help me how to make it work in my code? Where should I implement it?