Merge pull request #5533 from FederatedAI/dev-2.1.0-bq-fix

update bq Signed-off-by: weijingchen <talkingwallace@sohu.com>

examples/benchmark_quality/hetero_nn/hetero_nn.py

@@ -116,7 +116,7 @@ def main(config="../../config.yaml", param="./breast_config.yaml", namespace="")
         'eval_0',
         runtime_parties=dict(guest=guest),
         metrics=metrics,
-        input_data=[hetero_nn_0.outputs['train_output_data'], hetero_nn_1.outputs['test_output_data']]
+        input_datas=[hetero_nn_0.outputs['train_output_data'], hetero_nn_1.outputs['test_output_data']]
     )
 
     pipeline.add_tasks([reader_0, psi_0, hetero_nn_0, hetero_nn_1, evaluation_0])

examples/benchmark_quality/hetero_secureboost/pipeline_hetero_sbt.py

@@ -36,7 +36,7 @@ def main(config="../../config.yaml", param="./sbt_breast_config.yaml", namespace
 
     assert isinstance(param, dict)
 
-    guest_data_table = param.get("data_guest")run
+    guest_data_table = param.get("data_guest")
     host_data_table = param.get("data_host")
 
     guest_train_data = {"name": guest_data_table, "namespace": f"experiment{namespace}"}
@@ -69,7 +69,7 @@ def main(config="../../config.yaml", param="./sbt_breast_config.yaml", namespace
         evaluation_0 = Evaluation(
             'eval_0',
             runtime_parties=dict(guest=guest),
-            input_data=[hetero_sbt_0.outputs['train_output_data']],
+            input_datas=[hetero_sbt_0.outputs['train_output_data']],
             default_eval_setting='regression',
         )
 
@@ -79,7 +79,7 @@ def main(config="../../config.yaml", param="./sbt_breast_config.yaml", namespace
             'eval_0',
             runtime_parties=dict(guest=guest),
             metrics=['auc'],
-            input_data=[hetero_sbt_0.outputs['train_output_data']]
+            input_datas=[hetero_sbt_0.outputs['train_output_data']]
         )
 
     pipeline.add_tasks([reader_0, psi_0, hetero_sbt_0, evaluation_0])

examples/benchmark_quality/homo_nn/pipeline_nn_binary.py

@@ -101,7 +101,7 @@ def main(config="../../config.yaml", param="./fed_nn_breast_config.yaml", namesp
         'eval_0',
         runtime_parties=dict(guest=guest),
         metrics=['auc'],
-        input_data=[homo_nn_1.outputs['test_output_data'], homo_nn_0.outputs['train_output_data']]
+        input_datas=[homo_nn_1.outputs['test_output_data'], homo_nn_0.outputs['train_output_data']]
     )
 
     if config.task_cores:

examples/benchmark_quality/homo_nn/pipeline_nn_multi.py

@@ -13,15 +13,13 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 #
-
 import argparse
 from fate_test.utils import parse_summary_result
+from fate_client.pipeline.components.fate import Reader
 from fate_client.pipeline import FateFlowPipeline
-from fate_client.pipeline.interface import DataWarehouseChannel
 from fate_client.pipeline.utils import test_utils
 from fate_client.pipeline.components.fate.evaluation import Evaluation
 from fate_client.pipeline import FateFlowPipeline
-from fate_client.pipeline.interface import DataWarehouseChannel
 from fate_client.pipeline.components.fate.nn.torch import nn, optim
 from fate_client.pipeline.components.fate.nn.loader import ModelLoader
 from fate_client.pipeline.components.fate.homo_nn import HomoNN, get_config_of_default_runner
@@ -68,34 +66,45 @@ def main(config="../../config.yaml", param="", namespace=""):
         fed_args=FedAVGArguments(),
         task_type='multi'
         )
+
+    reader_0 = Reader("reader_0", runtime_parties=dict(guest=guest, host=host))
+    reader_0.guest.task_parameters(
+        namespace=guest_train_data['namespace'],
+        name=guest_train_data['name']
+    )
+    reader_0.hosts[0].task_parameters(
+        namespace=host_train_data['namespace'],
+        name=guest_train_data['name']
+    )
 
+    reader_1 = Reader("reader_1", runtime_parties=dict(guest=guest, host=host), namespace=test_data["namespace"], name=test_data["name"])
 
     homo_nn_0 = HomoNN(
         'nn_0',
-        runner_conf=conf
+        runner_conf=conf,
+        train_data=reader_0.outputs["output_data"]
     )
 
     homo_nn_1 = HomoNN(
         'nn_1',
-        test_data=DataWarehouseChannel(name=test_data["name"], namespace=test_data["namespace"]),
-        input_model=homo_nn_0.outputs['output_model']
+        input_model=homo_nn_0.outputs['output_model'],
+        test_data=reader_1.outputs["output_data"]
     )
 
-    homo_nn_0.guest.task_parameters(train_data=DataWarehouseChannel(name=guest_train_data["name"], namespace=guest_train_data["namespace"]))
-    homo_nn_0.hosts[0].task_parameters(train_data=DataWarehouseChannel(name=host_train_data["name"], namespace=host_train_data["namespace"]))
-
     evaluation_0 = Evaluation(
         'eval_0',
         default_eval_setting='multi',
         runtime_parties=dict(guest=guest),
-        input_data=[homo_nn_1.outputs['test_output_data'], homo_nn_0.outputs['train_output_data']]
+        input_datas=[homo_nn_1.outputs['test_output_data'], homo_nn_0.outputs['train_output_data']]
     )
 
     if config.task_cores:
         pipeline.conf.set("task_cores", config.task_cores)
     if config.timeout:
         pipeline.conf.set("timeout", config.timeout)
 
+    pipeline.add_task(reader_0)
+    pipeline.add_task(reader_1)
     pipeline.add_task(homo_nn_0)
     pipeline.add_task(homo_nn_1)
     pipeline.add_task(evaluation_0)

examples/benchmark_quality/homo_nn/pipeline_nn_regression.py

@@ -13,15 +13,13 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 #
-
 import argparse
 from fate_test.utils import parse_summary_result
+from fate_client.pipeline.components.fate import Reader
 from fate_client.pipeline import FateFlowPipeline
-from fate_client.pipeline.interface import DataWarehouseChannel
 from fate_client.pipeline.utils import test_utils
 from fate_client.pipeline.components.fate.evaluation import Evaluation
 from fate_client.pipeline import FateFlowPipeline
-from fate_client.pipeline.interface import DataWarehouseChannel
 from fate_client.pipeline.components.fate.nn.torch import nn, optim
 from fate_client.pipeline.components.fate.nn.torch.base import Sequential
 from fate_client.pipeline.components.fate.homo_nn import HomoNN, get_config_of_default_runner
@@ -72,33 +70,43 @@ def main(config="../../config.yaml", param="", namespace=""):
         task_type='regression'
         )
 
+    reader_0 = Reader("reader_0", runtime_parties=dict(guest=guest, host=host))
+    reader_0.guest.task_parameters(
+        namespace=guest_train_data['namespace'],
+        name=guest_train_data['name']
+    )
+    reader_0.hosts[0].task_parameters(
+        namespace=host_train_data['namespace'],
+        name=guest_train_data['name']
+    )
+    reader_1 = Reader("reader_1", runtime_parties=dict(guest=guest, host=host), namespace=test_data["namespace"], name=test_data["name"])
 
     homo_nn_0 = HomoNN(
         'nn_0',
-        runner_conf=conf
+        runner_conf=conf,
+        train_data=reader_0.outputs["output_data"]
     )
 
     homo_nn_1 = HomoNN(
         'nn_1',
-        test_data=DataWarehouseChannel(name=test_data["name"], namespace=test_data["namespace"]),
-        input_model=homo_nn_0.outputs['output_model']
+        input_model=homo_nn_0.outputs['output_model'],
+        test_data=reader_1.outputs["output_data"]
     )
 
-    homo_nn_0.guest.task_parameters(train_data=DataWarehouseChannel(name=guest_train_data["name"], namespace=guest_train_data["namespace"]))
-    homo_nn_0.hosts[0].task_parameters(train_data=DataWarehouseChannel(name=host_train_data["name"], namespace=host_train_data["namespace"]))
-
     evaluation_0 = Evaluation(
         'eval_0',
         default_eval_setting='regression',
         runtime_parties=dict(guest=guest),
-        input_data=[homo_nn_1.outputs['test_output_data'], homo_nn_0.outputs['train_output_data']]
+        input_datas=[homo_nn_1.outputs['test_output_data'], homo_nn_0.outputs['train_output_data']]
     )
 
     if config.task_cores:
         pipeline.conf.set("task_cores", config.task_cores)
     if config.timeout:
         pipeline.conf.set("timeout", config.timeout)
 
+    pipeline.add_task(reader_0)
+    pipeline.add_task(reader_1)
     pipeline.add_task(homo_nn_0)
     pipeline.add_task(homo_nn_1)
     pipeline.add_task(evaluation_0)

examples/benchmark_quality/linr/fate-linr.py

@@ -72,7 +72,7 @@ def main(config="../../config.yaml", param="./linr_config.yaml", namespace=""):
                               metrics=["r2_score",
                                        "mse",
                                        "rmse"],
-                              input_data=linr_0.outputs["train_output_data"])
+                              input_datas=linr_0.outputs["train_output_data"])
     pipeline.add_tasks([reader_0, psi_0, linr_0, evaluation_0])
 
     if config.task_cores:

examples/benchmark_quality/linr/fate-sshe-linr.py

@@ -72,7 +72,7 @@ def main(config="../../config.yaml", param="./linr_config.yaml", namespace=""):
                               metrics=["r2_score",
                                        "mse",
                                        "rmse"],
-                              input_data=linr_0.outputs["train_output_data"])
+                              input_datas=linr_0.outputs["train_output_data"])
     pipeline.add_tasks([reader_0, psi_0, linr_0, evaluation_0])
 
     if config.task_cores:

examples/benchmark_quality/lr/pipeline-lr-binary.py

@@ -72,7 +72,7 @@ def main(config="../../config.yaml", param="./breast_config.yaml", namespace="")
     evaluation_0 = Evaluation("evaluation_0",
                               runtime_parties=dict(guest=guest),
                               metrics=["auc", "binary_precision", "binary_accuracy", "binary_recall"],
-                              input_data=lr_0.outputs["train_output_data"])
+                              input_datas=lr_0.outputs["train_output_data"])
     pipeline.add_tasks([reader_0, psi_0, lr_0, lr_1, evaluation_0])
 
     if config.task_cores:

examples/benchmark_quality/lr/pipeline-lr-multi.py

@@ -70,7 +70,7 @@ def main(config="../../config.yaml", param="./vehicle_config.yaml", namespace=""
 
     evaluation_0 = Evaluation('evaluation_0',
                               runtime_parties=dict(guest=guest),
-                              input_data=lr_0.outputs["train_output_data"],
+                              input_datas=lr_0.outputs["train_output_data"],
                               predict_column_name='predict_result',
                               metrics=['multi_recall', 'multi_accuracy', 'multi_precision'])
     pipeline.add_tasks([reader_0, psi_0, lr_0, lr_1, evaluation_0])

examples/benchmark_quality/lr/pipeline-sshe-lr-binary.py

@@ -74,7 +74,7 @@ def main(config="../../config.yaml", param="./breast_config.yaml", namespace="")
     evaluation_0 = Evaluation("evaluation_0",
                               runtime_parties=dict(guest=guest),
                               metrics=["auc", "binary_precision", "binary_accuracy", "binary_recall"],
-                              input_data=lr_0.outputs["train_output_data"])
+                              input_datas=lr_0.outputs["train_output_data"])
     pipeline.add_tasks([reader_0, psi_0, lr_0, lr_1, evaluation_0])
 
     if config.task_cores:

examples/benchmark_quality/lr/pipeline-sshe-lr-multi.py

@@ -71,7 +71,7 @@ def main(config="../../config.yaml", param="./vehicle_config.yaml", namespace=""
 
     evaluation_0 = Evaluation('evaluation_0',
                               runtime_parties=dict(guest=guest),
-                              input_data=lr_0.outputs["train_output_data"],
+                              input_datas=lr_0.outputs["train_output_data"],
                               predict_column_name='predict_result',
                               metrics=['multi_recall', 'multi_accuracy', 'multi_precision'])
     pipeline.add_tasks([reader_0, psi_0, lr_0, lr_1, evaluation_0])