123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105 |
- import argparse
- from pipeline.backend.pipeline import PipeLine
- from pipeline.component import DataTransform
- from pipeline.component import Evaluation
- from pipeline.component import HeteroSSHELinR
- from pipeline.component import Intersection
- from pipeline.component import Reader
- from pipeline.component import SampleWeight
- from pipeline.interface import Data
- from pipeline.utils.tools import load_job_config
- def main(config="../../config.yaml", namespace=""):
-
- if isinstance(config, str):
- config = load_job_config(config)
- parties = config.parties
- guest = parties.guest[0]
- host = parties.host[0]
- guest_train_data = {"name": "motor_hetero_guest", "namespace": f"experiment{namespace}"}
- host_train_data = {"name": "motor_hetero_host", "namespace": f"experiment{namespace}"}
- pipeline = PipeLine().set_initiator(role='guest', party_id=guest).set_roles(guest=guest, host=host)
- reader_0 = Reader(name="reader_0")
- reader_0.get_party_instance(role='guest', party_id=guest).component_param(table=guest_train_data)
- reader_0.get_party_instance(role='host', party_id=host).component_param(table=host_train_data)
- data_transform_0 = DataTransform(name="data_transform_0")
- data_transform_0.get_party_instance(role='guest', party_id=guest).component_param(with_label=True,
- label_name="motor_speed",
- label_type="float",
- output_format="dense")
- data_transform_0.get_party_instance(role='host', party_id=host).component_param(with_label=False)
- intersection_0 = Intersection(name="intersection_0")
- sample_weight_0 = SampleWeight(name="sample_weight_0")
- sample_weight_0.get_party_instance(role='guest', party_id=guest).component_param(need_run=True,
- sample_weight_name="pm")
- sample_weight_0.get_party_instance(role='host', party_id=host).component_param(need_run=False)
- hetero_linr_0 = HeteroSSHELinR(name="hetero_linr_0", penalty="L2", optimizer="rmsprop", tol=0.001,
- alpha=0.01, max_iter=20, early_stop="weight_diff", batch_size=-1,
- learning_rate=0.15, decay=0.0, decay_sqrt=False,
- init_param={"init_method": "zeros"},
- reveal_every_iter=True,
- reveal_strategy="respectively"
- )
- evaluation_0 = Evaluation(name="evaluation_0", eval_type="regression", pos_label=1)
-
- pipeline.add_component(reader_0)
- pipeline.add_component(data_transform_0, data=Data(data=reader_0.output.data))
- pipeline.add_component(intersection_0, data=Data(data=data_transform_0.output.data))
- pipeline.add_component(sample_weight_0, data=Data(data=intersection_0.output.data))
- pipeline.add_component(hetero_linr_0, data=Data(train_data=sample_weight_0.output.data))
- pipeline.add_component(evaluation_0, data=Data(data=hetero_linr_0.output.data))
- pipeline.compile()
- pipeline.fit()
-
-
- pipeline.deploy_component([data_transform_0, intersection_0, hetero_linr_0])
- predict_pipeline = PipeLine()
-
- predict_pipeline.add_component(reader_0)
-
-
- predict_pipeline.add_component(pipeline,
- data=Data(
- predict_input={pipeline.data_transform_0.input.data: reader_0.output.data}))
-
- predict_pipeline.predict()
- if __name__ == "__main__":
- parser = argparse.ArgumentParser("PIPELINE DEMO")
- parser.add_argument("-config", type=str,
- help="config file")
- args = parser.parse_args()
- if args.config is not None:
- main(args.config)
- else:
- main()
|