模型调试

spark/featureEngSkTfRecord.py

@@ -737,6 +737,7 @@ def get_spark(appName):
     sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
     sparkConf.set("mapreduce.output.fileoutputformat.compress", False)
     sparkConf.set("mapreduce.map.output.compress", False)
+    sparkConf.set("mapred.output.compress", False)
     spark = (SparkSession
              .builder
              .config(conf=sparkConf)
@@ -1113,16 +1114,23 @@ if __name__ == '__main__':
 
         return tf_serialized
 
-    output_file = "file:///home/gmuser/train_samples"
-    output_file = "/strategy/train_samples_tfrecord"
+    output_file = "/data/files/wideAndDeep/train_samples.tfrecord"
 
-    os.system("hdfs dfs -rmr {output_file}".format(output_file = output_file))
-    train_samples.rdd.map(get_example_string).coalesce(1).saveAsTextFile(output_file)
+    train_samples = train_samples.rdd.map(get_example_string).collect()
+    train_samples.unpersist(True)
 
-    output_file = "file:///home/gmuser/eval_samples"
-    output_file = "/strategy/eval_samples_tfrecord"
-    os.system("hdfs dfs -rmr {output_file}".format(output_file=output_file))
-    test_samples.rdd.map(get_example_string).coalesce(1).saveAsTextFile(output_file)
+    writer = tf.io.TFRecordWriter(output_file)
+    for train_sample in train_samples:
+        writer.write(train_sample)
+    writer.close()
+
+    output_file = "/data/files/wideAndDeep/eval_samples.tfrecord"
+    test_samples = test_samples.rdd.map(get_example_string).collect()
+
+    writer = tf.io.TFRecordWriter(output_file)
+    for test_sample in test_samples:
+        writer.write(test_sample)
+    writer.close()
 
     print("训练数据写入 耗时s:{}".format(time.time() - write_time_start))