'commit'

ETL/Mars/Test.py

@@ -0,0 +1,18 @@
+import datetime
+import time
+
+
+# 黄海定义的输出信息的办法，带当前时间
+def logInfo(msg):
+    i = datetime.datetime.now()
+    print(" %s            %s" % (i, msg))
+
+
+if __name__ == '__main__':
+    cnt = 0
+    while True:
+        logInfo("我还活着！")
+        time.sleep(1)
+        cnt = cnt + 1
+        if cnt == 30:
+            break

ETL/Mars/Utils/MySQLHelper.py

@@ -0,0 +1,69 @@
+
+import pymysql.cursors
+
+
+class MySQLHelper:
+    myVersion = 0.1
+    # 配置
+    user = 'root'
+    password = 'DsideaL147258369'
+    host = '10.10.14.230'
+    port = 22066
+    db = 'test'
+
+    def __init__(self, host=host, port=port, db=db, user=user, password=password, charset="utf8"):
+        self.host = host
+        self.user = user
+        self.port = port
+        self.password = password
+        self.charset = charset
+        self.db = db
+
+        try:
+            self.conn = pymysql.connect(host=self.host, port=self.port, user=self.user, passwd=self.password,
+                                        db=self.db, charset=self.charset, cursorclass=pymysql.cursors.DictCursor)
+            self.cursor = self.conn.cursor()
+
+            # 设置执行时间为8小时
+            sql = "set session wait_timeout=288000"
+            self.cursor.execute(sql)
+            sql = "set session interactive_timeout=288000"
+            self.cursor.execute(sql)
+
+        except Exception as e:
+            print('MySql Error : %d %s' % (e.args[0], e.args[1]))
+
+    def query(self, sql: object) -> object:
+        try:
+            self.cursor.execute(sql)
+            result = self.cursor.fetchall()
+            return result
+        except Exception as e:
+            print('MySql Error: %s SQL: %s' % (e, sql))
+
+    def execute(self, sql):
+        try:
+            self.cursor.execute(sql)
+            self.conn.commit()
+            # 获取更新了多少条数据
+            return self.cursor.rowcount
+        except Exception as e:
+            print('MySql Error: %s SQL: %s' % (e, sql))
+
+    def executeWithPara(self, sql, params):
+        try:
+            self.cursor.execute(sql, params)
+            self.conn.commit()
+        except Exception as e:
+            print('MySql Error: %s SQL: %s' % (e, sql))
+
+    def executemany(self, sql, data):
+        try:
+            self.cursor.executemany(sql, data)
+            self.conn.commit()
+        except Exception as e:
+            print('MySql Error: %s SQL: %s' % (e, sql))
+
+    def close(self):
+        self.cursor.close()
+        self.conn.close()

ETL/Mars/pack.json

@@ -0,0 +1,20 @@
+{
+  "t_zy_score": [
+    {
+      "sql": "update t_zy_score as t1 left join t_base_person as t2  on t1.person_id=t2.person_id set t1.check_flag=-1 where t2.person_id is  null limit 100",
+      "memo": "检查人员ID不存在的情况"
+    },
+    {
+      "sql": "update t_zy_score as t1 left join t_zy_main as t2  on t1.zy_id=t2.zy_id set t1.zy_name=t2.zy_name where t1.zy_name is  null limit 100",
+      "memo": "扩展作业名称字段内容"
+    },
+    {
+      "sql": "update t_zy_score as t1 set t1.check_flag=-1 where t1.zy_name is null limit 100",
+      "memo": "没补上名称的就是坏人"
+    },
+    {
+      "sql": "update t_zy_score as t1 set check_flag=1 where check_flag=0 limit 100",
+      "memo": "打上成功标识"
+    }
+  ]
+}

ETL/Mars/pack.py

@@ -1,15 +1,20 @@
 # python -m pip install --upgrade pip -i http://pypi.douban.com/simple/
 # pip install pymysql -i http://pypi.douban.com/simple/
 import datetime
-import sys
+import json
 
 import pymysql.cursors
 
 
 class MySQLHelper:
-    myVersion = 0.1
+    # 配置
+    user = 'root'
+    password = 'DsideaL147258369'
+    host = '10.10.14.230'
+    port = 22066
+    db = 'huanghai'
 
-    def __init__(self, host, port, db, user, password, charset="utf8"):
+    def __init__(self, host=host, port=port, db=db, user=user, password=password, charset="utf8"):
         self.host = host
         self.user = user
         self.port = port
@@ -73,30 +78,23 @@ def logInfo(msg):
     print(" %s            %s" % (i, msg))
 
 
-# 准备扩展
-_dict = {
-    "t_resource_base": "update t_resource_base t1,t_dm_stage t2 set t1.CHECK_MESSAGE=t2.stage_name where t1.stage_id=t2.stage_id and t1.CHECK_MESSAGE is null limit 100"}
-
 if __name__ == '__main__':
-    if len(sys.argv) == 1:
+    with open("./pack.json", "r", encoding="utf-8") as f:
-        print("没有正确输入参数，请检查！输入样例: python pack.py t_resource_base")
+        _dict = json.load(f)
-        sys.exit()
+
-    tableName = sys.argv[1]
+    #
-
+    allcnt = 0
-    # 配置
+    db = MySQLHelper()
-    user = 'root'
+    for _bean in _dict:
-    password = 'DsideaL147258369'
+        actions = _dict[_bean]
-    host = '10.10.14.230'
+        for action in actions:
-    port = 22066
+            memo = action["memo"]
-    database = 'test'
+            sql = action["sql"]
-
+            while True:
-    db = MySQLHelper(user=user, host=host, port=port, password=password, db=database)
+                cnt = db.execute(sql)
-
+                allcnt = allcnt + cnt
-    while True:
+                if cnt == 0:
-        sql = _dict[tableName]
+                    break
-        cnt = db.execute(sql)
+                logInfo(memo + ",更新%s条" % cnt)
-        logInfo("成功更新%s条" % cnt)
-        if cnt == 0:
-            break
-
     db.close()
+    logInfo("恭喜，所有清洗工作成功完成，共清洗%s条" % allcnt)