Add data lag operation in the Runnable extract_ts_features (#84)

* Move db utils method to db_adapter.py

* Adjust the import order

* Add TODO comments of IO adapter in sqlflow runtime library.

* Add __init__ in run_io folder

* Add lag columns before rolling time series data and extract features.

* Rename Rfc1783 to Rfc1738

Author: brightcoder01

runnables/extract_ts_features.py

@@ -1,9 +1,9 @@
 import argparse
 import os
 import pandas as pd 
-from run_io.db_adapter import convertDSNToRfc1783
+from run_io.db_adapter import convertDSNToRfc1738
 from sqlalchemy import create_engine
-from time_series_processing.ts_feature_extractor import add_features_extracted_from_ts_data
+from time_series_processing.ts_feature_extractor import add_features_extracted_from_ts_data, add_lag_columns
 
 
 def build_argument_parser():
@@ -12,6 +12,7 @@ def build_argument_parser():
     parser.add_argument("--column_id", type=str, required=True)
     parser.add_argument("--column_time", type=str, required=True)
     parser.add_argument("--columns_value", type=str, required=True)
+    parser.add_argument("--lag_num", type=int, default=1)
     parser.add_argument("--windows", type=str, required=True)
     parser.add_argument("--min_window", type=str, default=0)
     parser.add_argument("--extract_setting", type=str, default="minimal", choices=["minimal", "efficient", "comprehensive"])
@@ -29,23 +30,27 @@ def build_argument_parser():
     output = os.getenv("SQLFLOW_TO_RUN_INTO")
     datasource = os.getenv("SQLFLOW_DATASOURCE")
 
-    url = convertDSNToRfc1783(datasource, args.dbname)
+    url = convertDSNToRfc1738(datasource, args.dbname)
     engine = create_engine(url)
     input = pd.read_sql(
         sql=select_input,
         con=engine)
 
+    df_with_lag_columns, lag_column_names = add_lag_columns(input, columns_value, args.lag_num)
+
     print("Start extracting the features from the time series data.")
     df_with_extracted_features = add_features_extracted_from_ts_data(
-        input,
+        df_with_lag_columns,
         column_id=args.column_id,
         column_time=args.column_time,
-        columns_value=columns_value,
+        columns_value=lag_column_names,
         windows=windows,
         min_window=args.min_window,
         extract_setting=args.extract_setting)
     print("Complete the feature extraction.")
 
+    df_with_extracted_features = df_with_extracted_features.drop(columns=lag_column_names)
+
     df_with_extracted_features.to_sql(
         name=output,
         con=engine,

runnables/run_io/db_adapter.py

@@ -15,7 +15,7 @@ def parseMySQLDSN(dsn):
 # TODO(brightcoder01): Should we put this kind of common method
 # in sqlflow runtime? While writing the runnable code, users can
 # import the runtime library.
-def convertDSNToRfc1783(driver_dsn, defaultDbName):
+def convertDSNToRfc1738(driver_dsn, defaultDbName):
     driver, dsn = driver_dsn.split("://")
     user, passwd, host, port, database, config = parseMySQLDSN(dsn)
 

runnables/time_series_processing/ts_feature_extractor.py

@@ -52,6 +52,19 @@ def _roll_ts_and_extract_features(
     return extracted_features
 
 
+def add_lag_columns(
+    input,
+    columns_value,
+    lag_num):
+    lag_column_names = []
+    for column_value in columns_value:
+        lag_column_name = "{}_lag_{}".format(column_value, lag_num)
+        input[lag_column_name] = input[column_value].shift(lag_num)
+        lag_column_names.append(lag_column_name)
+    
+    return input[lag_num:], lag_column_names
+
+
 def add_features_extracted_from_ts_data(
     input,
     column_id,