{"dag_id":"N1wP690lywNj1xrkzaghe","user_name":"XXX","user_id":1,"nodes_task_name":"dfs","nodes_task_id":123,"itermidate_data":["hdfs://host:port/uri"],"nodes":[{"id":"8CSbH3t1CTdxSKZ0xqHf3","name":"datasourceNode1","op":"datasource","data":{"input_source":[{"dataSelect":true,"dataField":"uuid","dataType":"string"},{"dataSelect":true,"dataField":"label","dataType":"int"},{"dataSelect":true,"dataField":"feature1","dataType":"double"},{"dataSelect":true,"dataField":"feature2","dataType":"double"},{"dataSelect":true,"dataField":"feature3","dataType":"double"},{"dataSelect":true,"dataField":"feature4","dataType":"double"},{"dataSelect":true,"dataField":"feature5","dataType":"double"},{"dataSelect":true,"dataField":"feature6","dataType":"double"},{"dataSelect":true,"dataField":"feature7","dataType":"double"},{"dataSelect":true,"dataField":"feature8","dataType":"double"},{"dataSelect":true,"dataField":"feature9","dataType":"double"}],"input_table":0}},{"id":"GqZitaihZUjpafezqtLOf","name":"test","op":"datasource","data":{"input_source":[{"dataSelect":true,"dataField":"uuid","dataType":"string"},{"dataSelect":true,"dataField":"label","dataType":"int"},{"dataSelect":true,"dataField":"feature1","dataType":"double"},{"dataSelect":true,"dataField":"feature2","dataType":"double"},{"dataSelect":true,"dataField":"feature3","dataType":"double"},{"dataSelect":true,"dataField":"feature4","dataType":"double"},{"dataSelect":true,"dataField":"feature5","dataType":"double"},{"dataSelect":true,"dataField":"feature6","dataType":"double"},{"dataSelect":true,"dataField":"feature7","dataType":"double"},{"dataSelect":true,"dataField":"feature8","dataType":"double"},{"dataSelect":true,"dataField":"feature9","dataType":"double"}],"input_table":1}},{"id":"SHb3zarKGtOvBFvcaKTKj","name":"pysparkNode1","op":"pyspark","data":{"input_number":2,"output":[{"outputVar":"r0"}],"script":"# import sys\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql.types import *\nfrom time import *\n\nfrom pyspark.ml.classification import LogisticRegression\nfrom pyspark.ml.feature import  VectorAssembler\nfrom pyspark.ml import Pipeline\nfrom pyspark.sql.functions import udf, col\n\ndef getFeatureName():\n    featureLst = ['feature1', 'feature2', 'feature3', 'feature4', 'feature5', 'feature6', 'feature7', 'feature8', 'feature9']\n    colLst = ['uuid', 'label'] + featureLst\n    return featureLst, colLst\n\ndef parseFloat(x):\n    try:\n        rx = float(x)\n    except:\n        rx = 0.0\n    return rx\n\n\ndef getDict(dictDataLst, colLst):\n    dictData = {}\n    for i in range(len(colLst)):\n        dictData[colLst[i]] = parseFloat(dictDataLst[i])\n    return dictData\n\ndef to_array(col):\n    def to_array_(v):\n        return v.toArray().tolist()\n    # Important: asNondeterministic requires Spark 2.3 or later\n    # It can be safely removed i.e.\n    # return udf(to_array_, ArrayType(DoubleType()))(col)\n    # but at the cost of decreased performance\n    return udf(to_array_, ArrayType(DoubleType())).asNondeterministic()(col)\n\n# def read_input(ss: SparkSession, t1, t2):\n#     trainDF = ss.sql(f'select * from {t1}')\n#     testDF = ss.sql(f'select * from {t2}')\n#     return trainDF, testDF\n\n# def write_output(df1, m1):\n#     df1.write.mode(\"overwrite\").saveAsTable('prediction')\n#     m1.write().overwrite().save(\"hdfs://192.168.199.27:9000/tmp/dag/target/model/lrModel\")\n\n\n\n\ndef main_func(input0, input1, spark, sc):\n    trainDF = input0\n    testDF = input1\n    \n\n    featureLst, colLst = getFeatureName()\n\n\n    #用于训练的特征featureLst\n    vectorAssembler = VectorAssembler().setInputCols(featureLst).setOutputCol(\"features\")\n\n    #### 训练 ####\n    print(\"step 1\")\n    lr = LogisticRegression(regParam=0.01, maxIter=100)  # regParam 正则项参数\n\n    pipeline = Pipeline(stages=[vectorAssembler, lr])\n    model = pipeline.fit(trainDF)\n    #打印参数\n    print(\"\\n-------------------------------------------------------------------------\")\n    print(\"LogisticRegression parameters:\\n\" + lr.explainParams() + \"\\n\")\n    print(\"-------------------------------------------------------------------------\\n\")\n\n    #### 预测, 保存结果 ####\n    print(\"step 2\")\n    labelsAndPreds = model.transform(testDF).withColumn(\"probability_xj\", to_array(col(\"probability\"))[1])\\\n                                            .select(\"uuid\", \"label\", \"prediction\", \"probability_xj\")\n    labelsAndPreds.show()\n    # labelsAndPreds.write.mode(\"overwrite\").options(header=\"true\").csv(outputHDFS + \"/target/output\")\n\n\n    #### 评估不同阈值下的准确率、召回率\n    print(\"step 3\")\n    labelsAndPreds_label_1 = labelsAndPreds.where(labelsAndPreds.label == 1)\n    labelsAndPreds_label_0 = labelsAndPreds.where(labelsAndPreds.label == 0)\n    labelsAndPreds_label_1.show(3)\n    labelsAndPreds_label_0.show(3)\n    t_cnt = labelsAndPreds_label_1.count()\n    f_cnt = labelsAndPreds_label_0.count()\n    print(\"thre\\ttp\\ttn\\tfp\\tfn\\taccuracy\\trecall\")\n    for thre in [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9]:\n        tp = labelsAndPreds_label_1.where(labelsAndPreds_label_1.probability_xj > thre).count()\n        tn = t_cnt - tp\n        fp = labelsAndPreds_label_0.where(labelsAndPreds_label_0.probability_xj > thre).count()\n        fn = f_cnt - fp\n        print(\"%.1f\\t%d\\t%d\\t%d\\t%d\\t%.4f\\t%.4f\"%(thre, tp, tn, fp, fn, float(tp)/(tp+fp), float(tp)/(t_cnt)))\n    \n    r0 = labelsAndPreds\n    return [r0]"}},{"id":"dkbV9oK3AZuaIljq_C9w3","name":"test1","op":"outputsource","data":{"output_source":[{"dataField":"abc","dataType":"int"}]}}],"edges":[{"source":"8CSbH3t1CTdxSKZ0xqHf3","target":"SHb3zarKGtOvBFvcaKTKj"},{"source":"GqZitaihZUjpafezqtLOf","target":"SHb3zarKGtOvBFvcaKTKj"},{"source":"SHb3zarKGtOvBFvcaKTKj","target":"dkbV9oK3AZuaIljq_C9w3"}],"graph":{"cells":[{"shape":"dag-edge","attrs":{"line":{"strokeDasharray":"","style":{"animation":""}}},"id":"73ef6d80-3587-4bd6-b5ac-e2a16e8bb3d2","zIndex":-1,"source":{"cell":"d4fb565a-83b3-4831-8180-4b2043b620a8","port":"bottomPort"},"target":{"cell":"1aa2f6c2-fdb3-4241-ae06-8f8f71d84e68","port":"input0"}},{"shape":"dag-edge","attrs":{"line":{"strokeDasharray":"","style":{"animation":""}}},"id":"bd1cf153-d0cb-48ff-a13d-0aca75295fc2","zIndex":-1,"source":{"cell":"55b8e5d3-ba4b-4ebc-b670-59951103e1e2","port":"bottomPort"},"target":{"cell":"1aa2f6c2-fdb3-4241-ae06-8f8f71d84e68","port":"input1"}},{"shape":"dag-edge","attrs":{"line":{"strokeDasharray":""}},"id":"35a1272b-cb76-4bef-86b4-e87ab418329c","zIndex":-1,"source":{"cell":"1aa2f6c2-fdb3-4241-ae06-8f8f71d84e68","port":"r0"},"target":{"cell":"c6d091b7-95a2-4195-8cf0-201e2f0e4db6","port":"topPort"}},{"position":{"x":170,"y":120},"size":{"width":180,"height":80},"view":"react-shape-view","shape":"dag-node","component":{"key":null,"ref":null,"props":{},"_owner":null,"_store":{}},"portMarkup":[{"tagName":"foreignObject","selector":"fo","children":[{"ns":"http://www.w3.org/1999/xhtml","tagName":"body","selector":"foBody","attrs":{"xmlns":"http://www.w3.org/1999/xhtml"},"style":{"width":"100%","height":"100%","background":"transparent"},"children":[{"tagName":"div","selector":"foContent","style":{"width":"100%","height":"100%"}}]}]}],"ports":{"groups":{"top":{"position":"top","attrs":{"fo":{"width":10,"height":10,"x":-5,"y":-5,"magnet":"true"}}},"bottom":{"position":"bottom","attrs":{"fo":{"width":10,"height":10,"x":-5,"y":-5,"magnet":"true"}}}},"items":[{"id":"bottomPort","group":"bottom"}]},"id":"d4fb565a-83b3-4831-8180-4b2043b620a8","data":{"label":"InputSource","status":"default","type":"datasource","id":"d4fb565a-83b3-4831-8180-4b2043b620a8","nodeId":"8CSbH3t1CTdxSKZ0xqHf3","inputSource":[{"dataSelect":true,"dataField":"uuid","dataType":"string"},{"dataSelect":true,"dataField":"label","dataType":"int"},{"dataSelect":true,"dataField":"feature1","dataType":"double"},{"dataSelect":true,"dataField":"feature2","dataType":"double"},{"dataSelect":true,"dataField":"feature3","dataType":"double"},{"dataSelect":true,"dataField":"feature4","dataType":"double"},{"dataSelect":true,"dataField":"feature5","dataType":"double"},{"dataSelect":true,"dataField":"feature6","dataType":"double"},{"dataSelect":true,"dataField":"feature7","dataType":"double"},{"dataSelect":true,"dataField":"feature8","dataType":"double"},{"dataSelect":true,"dataField":"feature9","dataType":"double"}],"dataTableId":0,"dataTableName":"train","nodeName":"datasourceNode1"},"zIndex":1},{"position":{"x":530,"y":120},"size":{"width":180,"height":80},"view":"react-shape-view","shape":"dag-node","component":{"key":null,"ref":null,"props":{},"_owner":null,"_store":{}},"portMarkup":[{"tagName":"foreignObject","selector":"fo","children":[{"ns":"http://www.w3.org/1999/xhtml","tagName":"body","selector":"foBody","attrs":{"xmlns":"http://www.w3.org/1999/xhtml"},"style":{"width":"100%","height":"100%","background":"transparent"},"children":[{"tagName":"div","selector":"foContent","style":{"width":"100%","height":"100%"}}]}]}],"ports":{"groups":{"top":{"position":"top","attrs":{"fo":{"width":10,"height":10,"x":-5,"y":-5,"magnet":"true"}}},"bottom":{"position":"bottom","attrs":{"fo":{"width":10,"height":10,"x":-5,"y":-5,"magnet":"true"}}}},"items":[{"id":"bottomPort","group":"bottom"}]},"id":"55b8e5d3-ba4b-4ebc-b670-59951103e1e2","data":{"label":"InputSource","status":"default","type":"datasource","id":"55b8e5d3-ba4b-4ebc-b670-59951103e1e2","nodeId":"GqZitaihZUjpafezqtLOf","inputSource":[{"dataSelect":true,"dataField":"uuid","dataType":"string"},{"dataSelect":true,"dataField":"label","dataType":"int"},{"dataSelect":true,"dataField":"feature1","dataType":"double"},{"dataSelect":true,"dataField":"feature2","dataType":"double"},{"dataSelect":true,"dataField":"feature3","dataType":"double"},{"dataSelect":true,"dataField":"feature4","dataType":"double"},{"dataSelect":true,"dataField":"feature5","dataType":"double"},{"dataSelect":true,"dataField":"feature6","dataType":"double"},{"dataSelect":true,"dataField":"feature7","dataType":"double"},{"dataSelect":true,"dataField":"feature8","dataType":"double"},{"dataSelect":true,"dataField":"feature9","dataType":"double"}],"dataTableId":1,"dataTableName":"test","nodeName":"test"},"zIndex":2},{"position":{"x":360,"y":440},"size":{"width":180,"height":36},"view":"react-shape-view","shape":"dag-node","component":{"key":null,"ref":null,"props":{},"_owner":null,"_store":{}},"portMarkup":[{"tagName":"foreignObject","selector":"fo","children":[{"ns":"http://www.w3.org/1999/xhtml","tagName":"body","selector":"foBody","attrs":{"xmlns":"http://www.w3.org/1999/xhtml"},"style":{"width":"100%","height":"100%","background":"transparent"},"children":[{"tagName":"div","selector":"foContent","style":{"width":"100%","height":"100%"}}]}]}],"ports":{"groups":{"top":{"position":"top","attrs":{"fo":{"width":10,"height":10,"x":-5,"y":-5,"magnet":"true"}}},"bottom":{"position":"bottom","attrs":{"fo":{"width":10,"height":10,"x":-5,"y":-5,"magnet":"true"}}}},"items":[{"group":"top","id":"input0"},{"group":"top","id":"input1"},{"group":"bottom","id":"r0"}]},"id":"1aa2f6c2-fdb3-4241-ae06-8f8f71d84e68","data":{"label":"pyspark","status":"success","type":"script","id":"1aa2f6c2-fdb3-4241-ae06-8f8f71d84e68","nodeId":"SHb3zarKGtOvBFvcaKTKj","nodeName":"pysparkNode1","scriptText":"# import sys\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql.types import *\nfrom time import *\n\nfrom pyspark.ml.classification import LogisticRegression\nfrom pyspark.ml.feature import  VectorAssembler\nfrom pyspark.ml import Pipeline\nfrom pyspark.sql.functions import udf, col\n\ndef getFeatureName():\n    featureLst = ['feature1', 'feature2', 'feature3', 'feature4', 'feature5', 'feature6', 'feature7', 'feature8', 'feature9']\n    colLst = ['uuid', 'label'] + featureLst\n    return featureLst, colLst\n\ndef parseFloat(x):\n    try:\n        rx = float(x)\n    except:\n        rx = 0.0\n    return rx\n\n\ndef getDict(dictDataLst, colLst):\n    dictData = {}\n    for i in range(len(colLst)):\n        dictData[colLst[i]] = parseFloat(dictDataLst[i])\n    return dictData\n\ndef to_array(col):\n    def to_array_(v):\n        return v.toArray().tolist()\n    # Important: asNondeterministic requires Spark 2.3 or later\n    # It can be safely removed i.e.\n    # return udf(to_array_, ArrayType(DoubleType()))(col)\n    # but at the cost of decreased performance\n    return udf(to_array_, ArrayType(DoubleType())).asNondeterministic()(col)\n\n# def read_input(ss: SparkSession, t1, t2):\n#     trainDF = ss.sql(f'select * from {t1}')\n#     testDF = ss.sql(f'select * from {t2}')\n#     return trainDF, testDF\n\n# def write_output(df1, m1):\n#     df1.write.mode(\"overwrite\").saveAsTable('prediction')\n#     m1.write().overwrite().save(\"hdfs://192.168.199.27:9000/tmp/dag/target/model/lrModel\")\n\n\n\n\ndef main_func(input0, input1, spark, sc):\n    trainDF = input0\n    testDF = input1\n    \n\n    featureLst, colLst = getFeatureName()\n\n\n    #用于训练的特征featureLst\n    vectorAssembler = VectorAssembler().setInputCols(featureLst).setOutputCol(\"features\")\n\n    #### 训练 ####\n    print(\"step 1\")\n    lr = LogisticRegression(regParam=0.01, maxIter=100)  # regParam 正则项参数\n\n    pipeline = Pipeline(stages=[vectorAssembler, lr])\n    model = pipeline.fit(trainDF)\n    #打印参数\n    print(\"\\n-------------------------------------------------------------------------\")\n    print(\"LogisticRegression parameters:\\n\" + lr.explainParams() + \"\\n\")\n    print(\"-------------------------------------------------------------------------\\n\")\n\n    #### 预测, 保存结果 ####\n    print(\"step 2\")\n    labelsAndPreds = model.transform(testDF).withColumn(\"probability_xj\", to_array(col(\"probability\"))[1])\\\n                                            .select(\"uuid\", \"label\", \"prediction\", \"probability_xj\")\n    labelsAndPreds.show()\n    # labelsAndPreds.write.mode(\"overwrite\").options(header=\"true\").csv(outputHDFS + \"/target/output\")\n\n\n    #### 评估不同阈值下的准确率、召回率\n    print(\"step 3\")\n    labelsAndPreds_label_1 = labelsAndPreds.where(labelsAndPreds.label == 1)\n    labelsAndPreds_label_0 = labelsAndPreds.where(labelsAndPreds.label == 0)\n    labelsAndPreds_label_1.show(3)\n    labelsAndPreds_label_0.show(3)\n    t_cnt = labelsAndPreds_label_1.count()\n    f_cnt = labelsAndPreds_label_0.count()\n    print(\"thre\\ttp\\ttn\\tfp\\tfn\\taccuracy\\trecall\")\n    for thre in [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9]:\n        tp = labelsAndPreds_label_1.where(labelsAndPreds_label_1.probability_xj > thre).count()\n        tn = t_cnt - tp\n        fp = labelsAndPreds_label_0.where(labelsAndPreds_label_0.probability_xj > thre).count()\n        fn = f_cnt - fp\n        print(\"%.1f\\t%d\\t%d\\t%d\\t%d\\t%.4f\\t%.4f\"%(thre, tp, tn, fp, fn, float(tp)/(tp+fp), float(tp)/(t_cnt)))\n    \n    r0 = labelsAndPreds\n    return [r0]","outputData":[{"outputVar":"r0"}],"inputNumber":2},"zIndex":5},{"position":{"x":360,"y":550},"size":{"width":180,"height":80},"view":"react-shape-view","shape":"dag-node","component":{"key":null,"ref":null,"props":{},"_owner":null,"_store":{}},"portMarkup":[{"tagName":"foreignObject","selector":"fo","children":[{"ns":"http://www.w3.org/1999/xhtml","tagName":"body","selector":"foBody","attrs":{"xmlns":"http://www.w3.org/1999/xhtml"},"style":{"width":"100%","height":"100%","background":"transparent"},"children":[{"tagName":"div","selector":"foContent","style":{"width":"100%","height":"100%"}}]}]}],"ports":{"groups":{"top":{"position":"top","attrs":{"fo":{"width":10,"height":10,"x":-5,"y":-5,"magnet":"true"}}},"bottom":{"position":"bottom","attrs":{"fo":{"width":10,"height":10,"x":-5,"y":-5,"magnet":"true"}}}},"items":[{"id":"topPort","group":"top"}]},"id":"c6d091b7-95a2-4195-8cf0-201e2f0e4db6","data":{"label":"OutputSource","status":"default","type":"outputsource","id":"c6d091b7-95a2-4195-8cf0-201e2f0e4db6","nodeId":"dkbV9oK3AZuaIljq_C9w3","nodeName":"test1","outputSource":[{"dataField":"abc","dataType":"int"}]},"zIndex":6}]}}