数据开发流程规范及数据监控

# coding=utf-8
import datetime
import sys
import os
import requests
import json
import pymysql

class test_monitor():

    def __init__(self):
        self.database='warehouse'
        self.host='localhost'
        self.username='test'
        self.password='123456'
        self.table_list = [
            {
                "table_name": "ods_tgc_scu_index_online",
                "ds": "date_create"
            },
            {
                "table_name": "mid_jx_order_detail_online_result",
                "ds": "created_at"
            }
        ]

    def get_data(self,table_name,ds_time):
        try:
            db = pymysql.connect(self.host, self.username, self.password, self.database, charset='utf8')
            cursor = db.cursor()
            yesterday = (datetime.datetime.now() - datetime.timedelta(days=1)).strftime("%Y-%m-%d")
            today = datetime.datetime.now().strftime("%Y-%m-%d")
            sql = 'select count(1) from '+ table_name + ' where ' + ds_time + ' >= %s and '+ ds_time + ' < %s'
            cursor.execute(sql,(yesterday,today))
            data = cursor.fetchone()
            if(data[0] > 0):
                num = data[0]
                return num
            else:
                num = 0
                return num
            cursor.close()
            db.close()
        except pymysql.InternalError as error:
            code, message = error.args
            print(">>>>>>>>>>>>>", code, message)
            return -1
    def push_data(self,table_name,result):
        day = (datetime.datetime.now() - datetime.timedelta(days=1)).strftime("%Y-%m-%d")
        db = pymysql.connect(self.host, self.username, self.password, self.database, charset='utf8')
        cursor = db.cursor()
        sql = 'INSERT INTO souche_enable_market_monitor (datab,table_name,num,ds) value (%s,%s,%s,%s)'
        cursor.execute(sql,(self.database,table_name,result,day))
        db.commit()
        cursor.close()
        db.close()
   

    def run(self):
        def dingmessage(self):
            webhook = "https://oapi.dingtalk.com/robot/send?access_token=XXXXXXXXX"
            header = {
                "Content-Type": "application/json",
                "Charset": "UTF-8"
            }
            message = {
                "msgtype": "text",
                "text": {
                    "content":self
                },
                "at": {
                "atMobiles": [   #此处为需要@什么人。填写具体用户
                    "此处为需要@什么人。填写具体用户",
                ],
                "isAtAll": True  #此处为是否@所有人 True  所有人  False  无需所有人
                }
            }
            message_json = json.dumps(message)
            info = requests.post(url=webhook,data=message_json,headers=header)
            print('发送成功')
            print(info.json())
        day = (datetime.datetime.now() - datetime.timedelta(days=1)).strftime("%Y-%m-%d") 
        for table in self.table_list:
            table_name = table['table_name']
            ds_time = table['ds']
            result = self.get_data(table_name,ds_time)
            self.push_data(table_name,result)
            dingmessage('监控报警:\n'+day+'\n'+table_name+"数据量为:"+str(result))


if __name__ == '__main__':
    test_monitor().run()

CREATE TABLE `warehouse.dj_rpt_check_conf` (
`db` varchar(8) NOT NULL COMMENT '数据库别名，例如bi，online,warehouse结果库)' ,
`tbl` varchar(64) NOT NULL COMMENT '表名',
`condition` varchar(256) NOT NULL COMMENT '筛选条件',
`threshold` bigint(20) NOT NULL DEFAULT 0 COMMENT '阈值',
`owner` varchar(16) NOT NULL default 'nobody' COMMENT '负责人：每个人自己固定用一个名字',
`ptype` varchar(8) NOT NULL COMMENT '检查周期,例如：d(天），w(周，周一）,m(月,1号)',
unique index tbl_db (tbl,db)
) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4;

插入数据：
insert into dj_rpt_check_conf values ('bi','dj_share_category_di','stat_date="${ds}"',20,'dy','d');

# coding: utf-8

from djtool import *
import pandas as pd 
import sqlalchemy as sq
from sqlalchemy import exc
import sys
import requests
import json
import copy

owner_mobile={'gw':'123456789'}

def check_table(conn,table,condition,threshold):
    try:
        cursor = conn.cursor()
        check_sql= 'select count(1) from ' + table + ' where ' + condition
        cursor.execute(check_sql)
        data = cursor.fetchone()
        if(data[0]<threshold):
            return data[0]
        else:
            return None
    except pymysql.InternalError as error:
        code, message = error.args
        print(">>>>>>>>>>>>>", code, message)
        return -1

ds= sys.argv[1]
check_conf = pd.read_sql_table('dj_rpt_check_conf',get_sqlalchemy_conn('mysql','bg'))

check_conf['condition'] = check_conf.condition.str.replace('\$\{ds\}',ds)
check_conf['real_cnt'] = 0
check_conf['failed'] = 0

for db in check_conf.db.unique():
    db_conn=get_pymysql_conn("mysql_"+db)
    for index,row in check_conf[(check_conf.db==db) & (check_conf.ptype=='d')].iterrows():
        real_cnt = check_table(db_conn,row['tbl'],row['condition'],row['threshold'])
        if(real_cnt is not None):
             check_conf.loc[index,'real_cnt']=real_cnt
             check_conf.loc[index,'failed']=1

mail_text = '''
配置表：`warehouse.dj_rpt_check_conf` 
'''
fail_conf = check_conf[(check_conf.failed==1)]

if(fail_conf.shape[0]>0):
    send_mail(['data@idongjia.cn'],[],ds+'--BI任务失败列表',mail_text + fail_conf.to_html())
else:
    send_mail(['data@idongjia.cn'],[],ds+'--已经加入监控的BI任务完成:)',mail_text)

headers = {'Content-Type': 'application/json'}
ding_url = 'https://oapi.dingtalk.com/robot/send?access_token=xxxxxxx'
msg={
     "msgtype": "markdown",
     "markdown": {"title":"BI任务失败了:"+ds,
"text":"#### BI任务失败了:"+ds+"  \n @mobile 失败任务：\n- fail_task "
     },
    "at": {
        "atMobiles": [
            "88888"
        ]
    }
 }

for owner in fail_conf.owner.unique():
    tmp_msg = copy.deepcopy(msg)
    tmp_msg['at']['atMobiles']=[owner_mobile[owner]]
    tmp_msg['markdown']['text']=tmp_msg['markdown']['text'].replace('mobile',owner_mobile[owner])
    tmp_msg['markdown']['text']=tmp_msg['markdown']['text'].replace('fail_task',"\n- ".join(fail_conf[(fail_conf.owner==owner)].tbl.values.tolist()))
    requests.post(ding_url, headers=headers,data=json.dumps(tmp_msg))

#!/usr/bin/python
# -*- coding: UTF-8 -*-
 
import smtplib
import sys
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
from email.utils import formataddr
from sys import argv

def main():
    sender = 'test'
    receivers = ['gaowei@test.cn']
    password = 'xxxxxxxx'
    message = MIMEMultipart()
    message['From'] = formataddr(["数据组",sender])
    message['To'] = formataddr(["数据组成员",receivers])
    subject = 'rest'
    message['Subject'] = Header(subject, 'utf-8')
    message.attach(MIMEText(sys.argv[1]+'数据见附件\n', 'plain', 'utf-8'))
    att1 = MIMEText(open("aa.csv", 'rb').read(), 'base64', 'utf-8')
    att1["Content-Type"] = 'application/octet-stream'
    att1["Content-Disposition"] = 'attachment; '+'filename='+sys.argv[1]+'.csv'
    message.attach(att1)
    try:
     server=smtplib.SMTP_SSL("smtp.exmail.qq.com", 465)
     server.login(sender, password)
     server.sendmail(sender,receivers,message.as_string())
     print "邮件发送成功"
    except smtplib.SMTPException:
     print "Error: 无法发送邮件"

if __name__ == '__main__':
    main()

#!/usr/bin/env bash
export JAVA_HOME=/opt/jdk1.8.0_121
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=$PATH:${JAVA_HOME}/bin:{JRE_HOME}/bin:$PATH
export PATH=$PATH:/opt/mysql/bin
dateStrDay=$1
if [ -z "$1" ] ; then
        dateStrDay=`date +%Y-%m-%d`
fi
echo $dateStrDay
dateStr=`date +%Y-%m-%d-%M-%S`

hadoop fs -rm -r  /tmp/gaowei/test

cd /opt/task/gaowei/warehouse/test
rm *.csv

spark2-submit --class cn.idongjia.data.auction.AuctionOrder --master yarn --deploy-mode cluster /opt/task/gaowei/warehouse/test/datawarehouse_2.11-1.0.jar

hadoop fs -getmerge /tmp/gaowei/test/* /opt/task/gaowei/warehouse/test/bb.csv

sed '1i\用户id,拍卖订单数,跑单数,异常订单数,是否禁言禁拍(0表示否，1表示是),是否在白名单(0表示否，1表示是),是否屏蔽(0表示否，1表示是,时间)' /opt/task/gaowei/warehouse/test/bb.csv > /opt/task/gaowei/warehouse/test/aa.csv

python Email.py ${dateStrDay}

package spark_tmp.utils
import java.io.File

import com.typesafe.config.ConfigFactory
import org.apache.spark.rdd.RDD
import play.api.libs.mailer._

object TaskSendMail {
  /**
    * 定义一个发邮件的人
    * @param host STMP服务地址
    * @param port STMP服务端口号
    * @param user STMP服务用户邮箱
    * @param password STMP服务邮箱密码
    * @param timeout setSocketTomeout 默认: 60s
    * @param connectionTimeout setSocketConnectionTimeout 默认：60s
    * @return  返回一个可以发邮件的用户
    */
  def createMailer(host:String, port: Int, user: String, password: String, timeout:Int = 10000, connectionTimeout:Int = 10000):SMTPMailer ={
    // STMP服务SMTPConfiguration
    val configuration = new SMTPConfiguration(
      host, port, false, false, false,
      Option(user), Option(password), false, timeout = Option(timeout),
      connectionTimeout = Option(connectionTimeout), ConfigFactory.empty(), false
    )
    val mailer: SMTPMailer = new SMTPMailer(configuration)
    mailer
  }


  /**
    * 生成一封邮件
    * @param subject 邮件主题
    * @param from 邮件发送地址
    * @param to 邮件接收地址
    * @param bodyText 邮件内容
    * @param bodyHtml 邮件的超文本内容
    * @param charset 字符编码 默认：utf-8
    * @param attachments 邮件的附件
    * @return 一封邮件
    */
  def createEmail(subject:String, from:String, to:Seq[String], bodyText:String = "ok", bodyHtml:String = "", charset:String = "utf-8", attachments:Seq[Attachment] = Seq.empty): Email = {

    val email = Email(subject, from, to,
      bodyText = Option[String](bodyText), bodyHtml = Option[String](bodyHtml),
      charset= Option[String](charset),attachments = attachments

    )
    email

  }

  /**
    *  生成一个附件
    * @param name 附件的名字
    * @param fileStr 以本地文件为附件相关参数
    * @param rdd 以hdfs文件或rdd或df为附件相关参数
    * @return
    */
  def createAttachments(name: String, fileStr: String = "", rdd:RDD[String] = null): Attachment  = {
    var attachment: Attachment = null
    if(fileStr.contains(":")){
      val file: File = new File(fileStr)
      attachment = AttachmentFile(name, file)
    }else{
      val data: Array[Byte] = rdd.collect().mkString("\n").getBytes()
      // 根据文件类型选择MimeTypes对应的值
      val mimetype = "text/plain"
      attachment = AttachmentData(name, data, mimetype)
    }
    attachment
  }


  /**
    *  主要针对日常简单结果的快速发送
    * @param subject 邮件主题名字
    * @param toStr 邮件的接收人，多名以,分割
    * @param bodyText 邮件的内容
    * @return 用户设备 <510109769.1.1561635225728@RAN>
    */
  def dailyEmail(subject:String, toStr:String, bodyText: String):String={
    val to = toStr.split(",").toList
    // 阿里云企业 邮箱
    val host = "smtp.189.cn"
    val port = 25
    val user = "18968044961@189.cn"
    val password = "xxxxxxxxxx"
    val from = user


    val mailer: SMTPMailer = TaskSendMail.createMailer(host, port, user, password)
    val email: Email = TaskSendMail.createEmail(subject, from, to, bodyText = bodyText)
    val userdev: String = mailer.send(email)
    userdev
  }
}