Remove hdfs params in Python (#2834)

* Remove hdfs params in Python

* fomat code

Author: lhw362950217

python/runtime/db.py

@@ -199,15 +199,7 @@ def buffered_db_writer(conn, table_name, table_schema, buff_size=100):
     elif driver == "mysql":
         w = db_writer.MySQLDBWriter(conn, table_name, table_schema, buff_size)
     elif driver == "hive":
-        w = db_writer.HiveDBWriter(
-            conn,
-            table_name,
-            table_schema,
-            buff_size,
-            hdfs_namenode_addr=conn.param("hdfs_namenode_addr", ""),
-            hive_location=conn.param("hive_location", ""),
-            hdfs_user=conn.param("hdfs_user", ""),
-            hdfs_pass=conn.param("hdfs_pass", ""))
+        w = db_writer.HiveDBWriter(conn, table_name, table_schema, buff_size)
     elif driver == "paiio":
         w = db_writer.PAIMaxComputeDBWriter(table_name, table_schema,
                                             buff_size)

python/runtime/db_writer/hive.py

@@ -21,23 +21,15 @@
 
 
 class HiveDBWriter(BufferedDBWriter):
-    def __init__(self,
-                 conn,
-                 table_name,
-                 table_schema,
-                 buff_size=10000,
-                 hdfs_namenode_addr="",
-                 hive_location="",
-                 hdfs_user="",
-                 hdfs_pass=""):
+    def __init__(self, conn, table_name, table_schema, buff_size=10000):
         super().__init__(conn, table_name, table_schema, buff_size)
         self.tmp_f = tempfile.NamedTemporaryFile(dir="./")
         self.f = open(self.tmp_f.name, "w")
         self.schema_idx = self._indexing_table_schema(table_schema)
-        self.hdfs_namenode_addr = hdfs_namenode_addr
-        self.hive_location = hive_location
-        self.hdfs_user = hdfs_user
-        self.hdfs_pass = hdfs_pass
+        self.hdfs_namenode_addr = conn.param("hdfs_namenode_addr")
+        self.hive_location = conn.param("hive_location")
+        self.hdfs_user = conn.uripts.username
+        self.hdfs_pass = conn.uripts.password
 
     def _indexing_table_schema(self, table_schema):
         column_list = self.conn.get_table_schema(self.table_name)

python/runtime/dbapi/hive.py

@@ -82,7 +82,8 @@ def __init__(self, conn_uri):
                                   if k.startswith("session.")])
 
     def _get_result_set(self, statement):
-        cursor = self._conn.cursor(configuration=self._session_cfg)
+        cursor = self._conn.cursor(user=self.uripts.username,
+                                   configuration=self._session_cfg)
         try:
             cursor.execute(statement.rstrip(";"))
             return HiveResultSet(cursor)

python/runtime/pai/tensorflow/evaluate.py

@@ -14,6 +14,7 @@
 import sys
 
 import tensorflow as tf
+from runtime.dbapi.paiio import PaiIOConnection
 from runtime.model import oss
 from runtime.pai.pai_distributed import define_tf_flags
 from runtime.tensorflow import is_tf_estimator
@@ -131,12 +132,5 @@ def _evaluate(datasource,
 
     if result_table:
         metric_name_list = ["loss"] + validation_metrics
-        write_result_metrics(result_metrics,
-                             metric_name_list,
-                             result_table,
-                             "paiio",
-                             None,
-                             hdfs_namenode_addr="",
-                             hive_location="",
-                             hdfs_user="",
-                             hdfs_pass="")
+        write_result_metrics(result_metrics, metric_name_list, result_table,
+                             PaiIOConnection.from_table(result_table))

python/runtime/pai/tensorflow/explain.py

@@ -113,21 +113,19 @@ def _input_fn():
         return dataset.batch(1).cache()
 
     estimator = init_model_with_feature_column(estimator_cls, model_params)
-    driver = "paiio"
     conn = PaiIOConnection.from_table(result_table) if result_table else None
     if estimator_cls in (tf.estimator.BoostedTreesClassifier,
                          tf.estimator.BoostedTreesRegressor):
         explain_boosted_trees(datasource, estimator, _input_fn, plot_type,
-                              result_table, feature_column_names, driver, conn,
-                              "", "", "", "", oss_dest, oss_ak, oss_sk,
-                              oss_endpoint, oss_bucket_name)
+                              result_table, feature_column_names, conn,
+                              oss_dest, oss_ak, oss_sk, oss_endpoint,
+                              oss_bucket_name)
     else:
         shap_dataset = pd.DataFrame(columns=feature_column_names)
         for i, (features, label) in enumerate(_input_fn()):
             shap_dataset.loc[i] = [
                 item.numpy()[0][0] for item in features.values()
             ]
         explain_dnns(datasource, estimator, shap_dataset, plot_type,
-                     result_table, feature_column_names, driver, conn, "", "",
-                     "", "", oss_dest, oss_ak, oss_sk, oss_endpoint,
-                     oss_bucket_name)
+                     result_table, feature_column_names, conn, oss_dest,
+                     oss_ak, oss_sk, oss_endpoint, oss_bucket_name)

python/runtime/pai/tensorflow/predict.py

@@ -97,7 +97,6 @@ def _predict(datasource,
     model_params.update(feature_columns)
     is_estimator = is_tf_estimator(estimator)
 
-    driver = "paiio"
     conn = PaiIOConnection.from_table(pai_table)
     selected_cols = db.selected_cols(conn, None)
     predict_generator = db.db_generator(conn, None)
@@ -107,42 +106,16 @@ def _predict(datasource,
             # functional model need field_metas parameter
             model_params["field_metas"] = feature_metas
         print("Start predicting using keras model...")
-        keras_predict(estimator,
-                      model_params,
-                      save,
-                      result_table,
-                      feature_column_names,
-                      feature_metas,
-                      train_label_name,
-                      result_col_name,
-                      driver,
-                      conn,
-                      predict_generator,
-                      selected_cols,
-                      hdfs_namenode_addr="",
-                      hive_location="",
-                      hdfs_user="",
-                      hdfs_pass="")
+        keras_predict(estimator, model_params, save, result_table,
+                      feature_column_names, feature_metas, train_label_name,
+                      result_col_name, conn, predict_generator, selected_cols)
     else:
         model_params['model_dir'] = save
         print("Start predicting using estimator model...")
-        estimator_predict(estimator,
-                          model_params,
-                          save,
-                          result_table,
-                          feature_column_names,
-                          feature_column_names_map,
-                          feature_columns,
-                          feature_metas,
-                          train_label_name,
-                          result_col_name,
-                          driver,
-                          conn,
-                          predict_generator,
-                          selected_cols,
-                          hdfs_namenode_addr="",
-                          hive_location="",
-                          hdfs_user="",
-                          hdfs_pass="")
+        estimator_predict(estimator, model_params, save, result_table,
+                          feature_column_names, feature_column_names_map,
+                          feature_columns, feature_metas, train_label_name,
+                          result_col_name, conn, predict_generator,
+                          selected_cols)
 
     print("Done predicting. Predict table : %s" % result_table)

python/runtime/pai/xgboost/evaluate.py

@@ -53,10 +53,6 @@ def evaluate(datasource, select, data_table, result_table, oss_model_path,
               label_meta=label_meta,
               result_table=result_table,
               validation_metrics=metrics,
-              hdfs_namenode_addr="",
-              hive_location="",
-              hdfs_user="",
-              hdfs_pass="",
               is_pai=True,
               pai_table=data_table,
               model_params=model_params,

python/runtime/pai/xgboost/explain.py

@@ -55,10 +55,6 @@ def explain(datasource, select, data_table, result_table, label_column,
         result_table=result_table,
         is_pai=True,
         pai_explain_table=data_table,
-        hdfs_namenode_addr="",
-        hive_location="",
-        hdfs_user="",
-        hdfs_pass="",
         # (TODO:lhw) save/load explain result storage info into/from FLAGS
         oss_dest="",
         oss_ak="",

python/runtime/pai/xgboost/predict.py

@@ -58,10 +58,6 @@ def predict(datasource, select, data_table, result_table, label_column,
          pred_label_meta=label_meta,
          result_table=result_table,
          is_pai=True,
-         hdfs_namenode_addr="",
-         hive_location="",
-         hdfs_user="",
-         hdfs_pass="",
          pai_table=data_table,
          model_params=model_params,
          train_params=train_params,

python/runtime/tensorflow/evaluate.py

@@ -36,11 +36,7 @@ def evaluate(datasource,
              save="",
              batch_size=1,
              validation_steps=None,
-             verbose=0,
-             hdfs_namenode_addr="",
-             hive_location="",
-             hdfs_user="",
-             hdfs_pass=""):
+             verbose=0):
     estimator_cls = import_model(estimator_string)
     is_estimator = is_tf_estimator(estimator_cls)
     set_log_level(verbose, is_estimator)
@@ -68,18 +64,10 @@ def evaluate(datasource,
 
     # write result metrics to a table
     conn = connect_with_data_source(datasource)
-    driver = conn.driver
     if result_table:
         metric_name_list = ["loss"] + validation_metrics
-        write_result_metrics(result_metrics,
-                             metric_name_list,
-                             result_table,
-                             driver,
-                             conn,
-                             hdfs_namenode_addr=hdfs_namenode_addr,
-                             hive_location=hive_location,
-                             hdfs_user=hdfs_user,
-                             hdfs_pass=hdfs_pass)
+        write_result_metrics(result_metrics, metric_name_list, result_table,
+                             conn)
 
 
 def estimator_evaluate(estimator, eval_dataset, validation_metrics):
@@ -152,9 +140,7 @@ def get_features(sample, label):
     return result_metrics
 
 
-def write_result_metrics(result_metrics, metric_name_list, result_table,
-                         driver, conn, hdfs_namenode_addr, hive_location,
-                         hdfs_user, hdfs_pass):
+def write_result_metrics(result_metrics, metric_name_list, result_table, conn):
     # NOTE: assume that the result table is already created with columns:
     # loss | metric_names ...
     column_names = metric_name_list