Spark+Kafka實時監控Oracle數據預警

目標： 監控Oracle某張記錄表，有新增數據則獲取表數據，並推送到微信企業。

流程： Kafka實時監控Oracle指定表，獲取該表操作信息（日志），使用Spark Structured Streaming消費Kafka，獲取數據后清洗后存入指定目錄，Python實時監控該目錄，提取文本里面數據並推送到微信。（Oracle一台服務器，Kafka及Spark在另外一台服務器）

架構： Oracle+Kafka+Spark Structured Streaming+Python

centos7
oracle 11g
apache-maven-3.6.3-bin.tar.gz
kafka-connect-oracle-master.zip
hadoop-2.7.1.tar.gz
kafka_2.11-2.4.1.tgz （scala版本必須與系統及連接spark的jar包一致，這里是2.11）
spark-2.4.0-bin-without-hadoop.tgz
spark-streaming-kafka-0-8_2.11-2.4.0.jar
Java 1.8
python 3.6

一、Oracle側

這邊設置比較簡單，使用SYS或者SYSTEM賬戶開啟歸檔日志及附加日志即可，一般實際工作出於數據安全考慮日志都會開啟狀態，故不再多贅述，有搭建及開啟問題可以隨時私信。

二、Kafka側

①配置maven，並添加進環境變量

#下載地址：http://maven.apache.org/download.cgi
#解壓  所有配置文件默認放在/usr/local路徑
tar xvf apache-maven-3.6.3-bin.tar.gz -C /usr/local/

#修改環境變量
vi /etc/profile

#加入下面內容
export MAVEN_HOME=/usr/local/apache-maven
export PATH=$PATH:${MAVEN_HOME}/bin

#刷新配置
source /ect/profile

②配置kafka-connect-oracle-master，config文件按oracle側信息配置，然后使用maven工具編譯。

#壓縮包下載地址：https://github.com/erdemcer/kafka-connect-oracle
#解壓 
unzip kafka-connect-oracle-master.zip 

#修改config下的配置文件
vi kafka-connect-oracle-master/config/OracleSourceConnector.properties 

#修改內容如下：
db.name.alias=dbserver   #oracle實例名稱：select instance_name from v$instance
tasks.max=1
topic=cdczztar    #kafka主體名稱
db.name=DBSERVER    #oracle服務器：select name from v$database;
db.hostname=192.168.81.159  #oracle服務器地址
db.port=1521  #oracle端口，一般默認1521
db.user=test  #數據庫用戶名
db.user.password=123456  #數據庫密碼
db.fetch.size=1
table.whitelist=LINHL.LHL_TEST  #需要監控的表名，可以使用*號監控所有，必須大寫
table.blacklist=  #不監控的表名，沒有為空，缺少該行會報錯
parse.dml.data=true
reset.offset=true
start.scn=
multitenant=false

#編譯 ,成功會有提示，並生成target文件夾
cd /usr/local/kafka-connect-oracle-master
mvn clean package

③解壓kafka，並放入前面master文件夾下的幾個jar包及配置文件

#解壓 下載地址：http://kafka.apache.org/downloads
tar xvf kafka_2.11-2.4.1.tgz -C /usr/local/

#改名
mv ./kafka_2.11-2.4.1 ./kafka

#復制配置文件
cp /usr/local/kafka-connect-oracle-master/target/kafka-connect-oracle-1.0.71.jar  /usr/local/kafka/libs/
cp /usr/local/kafka-connect-oracle-master/lib/ojdbc7.jar  /usr/local/kafka/libs/
cp /usr/local/kafka-connect-oracle-master/config/OracleSourceConnector.properties  /usr/local/kafka/config/

④開啟Kafka

#進入Kafka文件夾
cd /usr/local/kafka/bin/

#下面全都在單獨的窗口開啟服務，勿關閉窗口，測試狀態，故沒有在后台運行
#啟動zookeeper
./zookeeper-server-start.sh ../config/zookeeper.properties
#啟動kafka服務
./kafka-server-start.sh ../config/server.properties
#建立topic-cdczztar
./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic cdczztar
#查看所有topic
./kafka-topics.sh --zookeeper localhost:2181 --list
#啟動連接oracle
./connect-standalone.sh ../config/connect-standalone.properties ../config/OracleSourceConnector.properties
#啟動消費端
#消費端此處只是為了展示用，后續使用spark做消費端
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic cdczztar

三、Spark側

Structured Streaming需要啟用HDFS，這里都在本地測試環境實現，因此關於java及hadoop的安裝，可以參考這篇的偽分布式配置dblab.xmu.edu.cn/blog/install-hadoop
①配置

#解壓 #官網可以下載，沒有資源請私信
tar  -zxf  spark-2.4.0-bin-without-hadoop.tgz  -C  /usr/local/

#重命名
mv  ./spark-2.4.0-bin-without-hadoop  ./spark

#修改配置文件
cd  /usr/local/spark
cp  ./conf/spark-env.sh.template  ./conf/spark-env.sh
vi ./conf/spark-env.sh

#加入下面內容
export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath):/usr/local/spark/examples/jars/*:/usr/local/spark/jars/kafka/*:/usr/local/kafka/libs/*

#修改系統環境變量
vi /etc/profile

#加入下面內容
export HADOOP_HOME=/usr/local/hadoop
export HADOOP_COMMON_LIB_NATIVE_DIR=$HADOOP_HOME/lib/native
export JAVA_HOME=/opt/java/jdk1.8.0_261
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=$PATH:${JAVA_HOME}/bin:/usr/local/hbase/bin
export SPARK_HOME=/usr/local/spark
export PYTHONPATH=$SPARK_HOME/python:$SPARK_HOME/python/lib/py4j-0.10.7-src.zip:/usr/local/python3/lib/python3.6/site-packages/:$PYTHONPATH
export PYSPARK_PYTHON=python3
export PATH=$HADOOP_HOME/bin:$SPARK_HOME/bin:$PATH

#更新配置
source /etc/profile

#在jars目錄建立kafka文件夾，把kafka所有jar包放到該目錄
cp /usr/local/spark-streaming-kafka-0-8_2.11-2.4.0.jar /usr/local/spark/jars/kafka
cp /usr/local/kafka/libs/* /usr/local/spark/jars/kafka

②Structured Streaming腳本建立

#!/usr/bin/env python3

import re
from functools import partial
from pyspark.sql.functions import *
from pyspark.sql import SparkSession

if __name__ == "__main__":

    spark = SparkSession \
        .builder \
        .appName("StructuredKafkaWordCount") \
        .getOrCreate()

    spark.sparkContext.setLogLevel('WARN') #只提示警示信息

    lines = spark \    #使用spark streaming則是基於KakfkaUtils包使用createDirectStream
        .readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", "localhost:9092") \
        .option("subscribe", 'cdczztar') \   #要消費的topic
        .load().selectExpr("CAST(value AS STRING)")

    #lines.printSchema()
    #正則處理，根據實際數據處理，kafka獲取后是oracle日志，在這只提取表插入的值
    pattern = 'data":(.+)}'
    fields = partial(regexp_extract, str="value", pattern=pattern)

    words = lines.select(fields(idx=1).alias("values"))
 
#輸出模式：存入文件
    query = words \
		.writeStream \
		.outputMode("append") \
		.format("csv") \
		.option("path","file:///tmp/filesink") \   #存到服務器地址
                .option("checkpointLocation","file:///tmp/file-sink-cp") \
		.trigger(processingTime="10 seconds") \
		.start()	
	
    query.awaitTermination()	


#新開一個服務器窗口運行，這邊已經在代碼目錄下
/usr/local/spark/bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.0  spark.py

③運行python實時打開寫入的文件，提取信息並推送到微信端

import csv
import pyinotify  #這個包只支持linux，如果是window系統可以使用watchdog，一個原理及寫法
import time
import requests
import json
import datetime
import pandas as pd

CORPID = "******"   #企業微信id
SECRET = "*******"  #企業微信密鑰
AGENTID = 1000041   #企業微信端口
multi_event = pyinotify.IN_CREATE  #只對create這個動作做監控
wm = pyinotify.WatchManager()

#繼承ProcessEvent后，對process_IN_CREATE方法重寫
class MyHandler(pyinotify.ProcessEvent):   

    def send_msg_to_wechat(self, content):
        record = '{}\n'.format(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
        s = requests.session()
        url1 = "https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={0}&corpsecret={1}".format(CORPID, SECRET)
        rep = s.get(url1)
        record += "{}\n".format(json.loads(rep.content))
        if rep.status_code == 200:
            token = json.loads(rep.content)['access_token']
            record += "獲取token成功\n"
        else:
            record += "獲取token失敗\n"
            token = None
        url2 = "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={}".format(token)
        header = {
            "Content-Type": "application/json"
        }
        form_data = {
            "touser": "@all",
            "toparty": " PartyID1 | PartyID2 ",
            "totag": " TagID1 | TagID2 ",
            "msgtype": "text",
            "agentid": AGENTID,
            "text": {
                "content": content
            },
            "safe": 0
        }
        rep = s.post(url2, data=json.dumps(form_data).encode('utf-8'), headers=header)
        if rep.status_code == 200:
            res = json.loads(rep.content)
            record += "發送成功\n"
        else:
            record += "發送失敗\n"
            res = None
        return res

    def process_IN_CREATE(self, event):

        try:
            if '_spark_metadata' in event.pathname or '.crc' in event.pathname:
                pass
            else:
                print(event.pathname)
                f_path = event.pathname
                #此處坑，streaming那邊生成文件還沒寫入數據就會觸發該任務，不sleep打開的是空白文件
                time.sleep(5)
                df = pd.read_csv(r'' + f_path, encoding='utf8', names=['value'], sep='/')
                send_str = df.iloc[0, 0].replace('\\', '').replace(',"before":null}', '').replace('"','')
                print(send_str)
                self.send_msg_to_wechat('中間庫預警：' + send_str)
        except:
            pass

handler = MyHandler()
notifier = pyinotify.Notifier(wm,handler)

wm.add_watch('/tmp/filesink/',multi_event)
notifier.loop()

微信端消息如下：

四、問題點

還有下面幾個問題還沒實現，有思路還請隨時評論私信交流，感謝

• 在structured streaming消費了kafka信息后，是否可以直接把消息推送到微信端口？

• python監控文件有新增文件路徑可以即時獲取，但是要獲取內容需要等待數據寫入，sleep的方式不穩定，是否有方法可以判斷數據已經寫完就讀取該文件？

---

學習交流，有任何問題還請隨時評論指出交流。