Spark+Kafka實時監控Oracle數據預警
目標： 監控Oracle某張記錄表，有新增數據則獲取表數據，並推送到微信企業。java
流程： Kafka實時監控Oracle指定表，獲取該表操做信息（日誌），使用Spark Structured Streaming消費Kafka，獲取數據後清洗後存入指定目錄，Python實時監控該目錄，提取文本里面數據並推送到微信。（Oracle一臺服務器，Kafka及Spark在另一臺服務器）python
架構： Oracle+Kafka+Spark Structured Streaming+Python
linux
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.6git
1、Oracle側github
這邊設置比較簡單，使用SYS或者SYSTEM帳戶開啓歸檔日誌及附加日誌便可，通常實際工做出於數據安全考慮日誌都會開啓狀態，故再也不多贅述，有搭建及開啓問題能夠隨時私信。sql
2、Kafka側數據庫
①配置maven，並添加進環境變量apache
#下載地址：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工具編譯。json
#壓縮包下載地址：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包及配置文件bootstrap
#解壓 下載地址：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
3、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()
微信端消息以下：
4、問題點
還有下面幾個問題還沒實現，有思路還請隨時評論私信交流，感謝
-
在structured streaming消費了kafka信息後，是否能夠直接把消息推送到微信端口？
-
python監控文件有新增文件路徑能夠即時獲取，可是要獲取內容須要等待數據寫入，sleep的方式不穩定，是否有方法能夠判斷數據已經寫完就讀取該文件？
學習交流，有任何問題還請隨時評論指出交流。
相關文章
- 1. [Oracle數據庫監控] Oracle告警警日誌文件、監聽監控腳本
- 2. 數據源監控與預警
- 3. GPE監控預警系統使用【 mysql數據庫監控】
- 4. 利用數據庫郵件服務實現監控和預警
- 5. 監控系統之預警
- 6. MySQL監控和預警
- 7. Redis監控和預警
- 8. oracle實時監控工具
- 9. 大數據實時案例--實時日誌監控告警系統
- 10. 應用監控預警--Pinpoint介紹
- 更多相關文章...
- • PHP 實例 - AJAX 實時搜索 - PHP教程
- • Redis watch命令——監控事務 - Redis教程
- • Flink 數據傳輸及反壓詳解
- • TiDB 在摩拜單車在線數據業務的應用和實踐
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. [Oracle數據庫監控] Oracle告警警日誌文件、監聽監控腳本
- 2. 數據源監控與預警
- 3. GPE監控預警系統使用【 mysql數據庫監控】
- 4. 利用數據庫郵件服務實現監控和預警
- 5. 監控系統之預警
- 6. MySQL監控和預警
- 7. Redis監控和預警
- 8. oracle實時監控工具
- 9. 大數據實時案例--實時日誌監控告警系統
- 10. 應用監控預警--Pinpoint介紹