update

output/out_put_diary_0923.py

@@ -29,39 +29,39 @@ from pyspark.sql import SparkSession, DataFrame
 # from pyspark.sql.functions import lit
 # import pytispark.pytispark as pti
 startTime = time.time()
-sparkConf = SparkConf()
-sparkConf.set("spark.sql.crossJoin.enabled", True)
-sparkConf.set("spark.debug.maxToStringFields", "100")
-sparkConf.set("spark.tispark.plan.allow_index_double_read", False)
-sparkConf.set("spark.tispark.plan.allow_index_read", True)
-sparkConf.set("spark.hive.mapred.supports.subdirectories", True)
-sparkConf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive", True)
-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("prod.gold.jdbcuri",
-              "jdbc:mysql://172.16.30.136/doris_prod?user=doris&password=o5gbA27hXHHm&rewriteBatchedStatements=true")
-sparkConf.set("prod.mimas.jdbcuri",
-              "jdbc:mysql://172.16.30.138/mimas_prod?user=mimas&password=GJL3UJe1Ck9ggL6aKnZCq4cRvM&rewriteBatchedStatements=true")
-sparkConf.set("prod.gaia.jdbcuri",
-              "jdbc:mysql://172.16.30.143/zhengxing?user=work&password=BJQaT9VzDcuPBqkd&rewriteBatchedStatements=true")
-sparkConf.set("prod.tidb.jdbcuri",
-              "jdbc:mysql://172.16.40.158:4000/eagle?user=st_user&password=aqpuBLYzEV7tML5RPsN1pntUzFy&rewriteBatchedStatements=true")
-sparkConf.set("prod.jerry.jdbcuri",
-              "jdbc:mysql://172.16.40.158:4000/jerry_prod?user=st_user&password=aqpuBLYzEV7tML5RPsN1pntUzFy&rewriteBatchedStatements=true")
-sparkConf.set("prod.tispark.pd.addresses", "172.16.40.158:2379")
-sparkConf.set("prod.tispark.pd.addresses", "172.16.40.170:4000")
-sparkConf.set("prod.tidb.database", "jerry_prod")
-sparkConf.setAppName("test")
-
-spark = (SparkSession.builder.config(conf=sparkConf).config("spark.sql.extensions", "org.apache.spark.sql.TiExtensions")
-         .config("spark.tispark.pd.addresses", "172.16.40.170:2379").enableHiveSupport().getOrCreate())
-
-spark.sql("ADD JAR hdfs:///user/hive/share/lib/udf/brickhouse-0.7.1-SNAPSHOT.jar")
-spark.sql("ADD JAR hdfs:///user/hive/share/lib/udf/hive-udf-1.0-SNAPSHOT.jar")
-spark.sql("CREATE TEMPORARY FUNCTION json_map AS 'brickhouse.udf.json.JsonMapUDF'")
-spark.sql("CREATE TEMPORARY FUNCTION is_json AS 'com.gmei.hive.common.udf.UDFJsonFormatCheck'")
-spark.sql("CREATE TEMPORARY FUNCTION arrayMerge AS 'com.gmei.hive.common.udf.UDFArryMerge'")
+# sparkConf = SparkConf()
+# sparkConf.set("spark.sql.crossJoin.enabled", True)
+# sparkConf.set("spark.debug.maxToStringFields", "100")
+# sparkConf.set("spark.tispark.plan.allow_index_double_read", False)
+# sparkConf.set("spark.tispark.plan.allow_index_read", True)
+# sparkConf.set("spark.hive.mapred.supports.subdirectories", True)
+# sparkConf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive", True)
+# 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("prod.gold.jdbcuri",
+#               "jdbc:mysql://172.16.30.136/doris_prod?user=doris&password=o5gbA27hXHHm&rewriteBatchedStatements=true")
+# sparkConf.set("prod.mimas.jdbcuri",
+#               "jdbc:mysql://172.16.30.138/mimas_prod?user=mimas&password=GJL3UJe1Ck9ggL6aKnZCq4cRvM&rewriteBatchedStatements=true")
+# sparkConf.set("prod.gaia.jdbcuri",
+#               "jdbc:mysql://172.16.30.143/zhengxing?user=work&password=BJQaT9VzDcuPBqkd&rewriteBatchedStatements=true")
+# sparkConf.set("prod.tidb.jdbcuri",
+#               "jdbc:mysql://172.16.40.158:4000/eagle?user=st_user&password=aqpuBLYzEV7tML5RPsN1pntUzFy&rewriteBatchedStatements=true")
+# sparkConf.set("prod.jerry.jdbcuri",
+#               "jdbc:mysql://172.16.40.158:4000/jerry_prod?user=st_user&password=aqpuBLYzEV7tML5RPsN1pntUzFy&rewriteBatchedStatements=true")
+# sparkConf.set("prod.tispark.pd.addresses", "172.16.40.158:2379")
+# sparkConf.set("prod.tispark.pd.addresses", "172.16.40.170:4000")
+# sparkConf.set("prod.tidb.database", "jerry_prod")
+# sparkConf.setAppName("test")
+#
+# spark = (SparkSession.builder.config(conf=sparkConf).config("spark.sql.extensions", "org.apache.spark.sql.TiExtensions")
+#          .config("spark.tispark.pd.addresses", "172.16.40.170:2379").enableHiveSupport().getOrCreate())
+#
+# spark.sql("ADD JAR hdfs:///user/hive/share/lib/udf/brickhouse-0.7.1-SNAPSHOT.jar")
+# spark.sql("ADD JAR hdfs:///user/hive/share/lib/udf/hive-udf-1.0-SNAPSHOT.jar")
+# spark.sql("CREATE TEMPORARY FUNCTION json_map AS 'brickhouse.udf.json.JsonMapUDF'")
+# spark.sql("CREATE TEMPORARY FUNCTION is_json AS 'com.gmei.hive.common.udf.UDFJsonFormatCheck'")
+# spark.sql("CREATE TEMPORARY FUNCTION arrayMerge AS 'com.gmei.hive.common.udf.UDFArryMerge'")
 
 task_list = []
 task_days = 2
@@ -98,13 +98,13 @@ today_str_format = now.strftime("%Y-%m-%d")
 yesterday_str = (now + datetime.timedelta(days=-1)).strftime("%Y%m%d")
 yesterday_str_format = (now + datetime.timedelta(days=-1)).strftime("%Y-%m-%d")
 one_week_age_str = (now + datetime.timedelta(days=-7)).strftime("%Y%m%d")
-sql = """select first_device from online.ml_user_history_detail where partition_date = {today_str} and last_active_date >= {last_30_day_str}
-""".format(today_str=today_str,last_30_day_str=last_30_day_str)
-print(sql)
-new_urser_device_id_df = spark.sql(sql)
-new_urser_device_id_df.createOrReplaceTempView("device_id_view")
-new_urser_device_id_df.show(1)
-sql_res = new_urser_device_id_df.collect()
+# sql = """select first_device from online.ml_user_history_detail where partition_date = {today_str} and last_active_date >= {last_30_day_str}
+# """.format(today_str=today_str,last_30_day_str=last_30_day_str)
+# print(sql)
+# new_urser_device_id_df = spark.sql(sql)
+# new_urser_device_id_df.createOrReplaceTempView("device_id_view")
+# new_urser_device_id_df.show(1)
+# sql_res = new_urser_device_id_df.collect()
 
 bulk_dict = {
     0: [0, 0, 0],