基于OracleStreamingService的IoT数据实时分析（第三弹）

# Make a Directory
mkdir -p connect-cli
cd connect-cli


# Download the CLI
wget https://github.com/lensesio/kafka-connect-tools/releases/download/v1.0.6/connect-cli


# Prepare Environment
# Load-Balancer-Public-IP:
kubectl get svc

# 拷贝上述 EXTERNAL-IP
130.61.197.91
export KAFKA_CONNECT_REST="http://"

name=mqtt-source
connector.class=com.datamountaineer.streamreactor.connect.mqtt.source.MqttSourceConnector
tasks.max=3
topics=iot-topic-stream
connect.mqtt.clean=true
connect.mqtt.timeout=1000
connect.mqtt.kcql=INSERT INTO iot-topic-stream SELECT * FROM mqtt-iot-topic WITHCOnVERTER=`com.datamountaineer.streamreactor.connect
connect.mqtt.keep.alive=1000
connect.mqtt.converter.throw.on.error=true
connect.mqtt.hosts=tcp://broker.hivemq.com:1883
connect.mqtt.service.quality=1
connect.progress.enabled=true

# connect-cli 权限
chmod 755 connect-cli


# 安装 Java
sudo yum install jre

./connect-cli create mqtt-source
./connect-cli ps

# mqtt-source 状态确认
./connect-cli status mqtt-source


connectorState: RUNNING
workerId: 10.244.0.12:8083
numberOfTasks: 1
tasks:
- taskId: 0
taskState: RUNNING
workerId: 10.244.0.12:8083


# 删除命令（必要时进行删除）
./connect-cli rm mqtt-source

sudo easy_install pip

sudo pip install paho-mqtt

# MQTT-Generator 代码下载
# git 未安装时需要进行安装> sudo yum install -y git
git clone https://github.com/ufthelp/MQTT-Generator.git


# 确认 config
cat config.json

python mqttgen.py 10 1 1

scp Wallet_test.zip opc@140.238.14.154:~

sudo su
wallet_unzipped_folder=/oracle_credentials_wallet
mkdir -p $wallet_unzipped_folder
unzip -u /Wallet_test.zip -d $wallet_unzipped_folder

docker cp oracle_credentials_wallet 0a7e6a8b8bcc:/oracle_credentials_wallet

docker start 0a7e6a8b8bcc
docker exec -it 0a7e6a8b8bcc /bin/bash

docker start 0a7e6a8b8bcc
docker exec -it 0a7e6a8b8bcc /bin/bash

confluent-hub install confluentinc/kafka-connect-jdbc:5.4.0

wget https://objectstorage.us-phoenix-1.oraclecloud.com/n/intmahesht/b/oracledrivers/o/ojdbc8-full.tar.gz
tar xvzf ./ojdbc8-full.tar.gz
cp ojdbc8-full/ojdbc8.jar /usr/share/java/kafka-connect-jdbc
rm -rf ojdbc8-full
rm -rf ojdbc8-full.tar.g

docker login icn.ocir.io
username:frvvnnkmcble/oracleidentitycloudservice/alvin.jin@oracle.com
password(Authtoken):xZ#89)pl{Mt5#gw


docker commit 0a7e6a8b8bcc icn.ocir.io/frvvnnkmcble/kafka-connect-mqtt/kafka-connect:0.2
docker push icn.ocir.io/frvvnnkmcble/kafka-connect-mqtt/kafka-connect:0.2

name=jdbc-adw
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
topics=iot-topic-stream
connection.url=jdbc:oracle:thin:@DBkafka_HIGH?TNS_ADMIN=/oracle_credentials_wallet
connection.user=ADMIN
connection.password=QWERasdf1234
auto.create=true

%script


-- Drop Previous Training Table
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE ML_AIRDATA';
EXCEPTION
WHEN OTHERS THEN NULL;
END;
/


create table ML_AIRDATA
as select a.sn, a.radon, a.co2, a.dust10, a.temperature, a.humidity, a.air_station_time, substr(a.air_station_time,1,15) time_id， b.hospital_no from admin.air_station a left join admin.air_station_master b on a.sn = b.sn;


select * from ml_airdata;

%script


-- Cleanup old settings table
BEGIN EXECUTE IMMEDIATE 'DROP TABLE esm_radon_settings';
EXCEPTION WHEN OTHERS THEN NULL; END;
/


-- Cleanup old model with the same name
BEGIN DBMS_DATA_MINING.DROP_MODEL('ESM_RADON_SAMPLE');
EXCEPTION WHEN OTHERS THEN NULL; END;
/


-- Create input time series
CREATE OR REPLACE VIEW ESM_RADON_DATA
AS SELECT AIR_STATION_TIME, RADON
FROM ML_AIRDATA where sn = 'FS301S000300';


CREATE TABLE ESM_RADON_SETTINGS(SETTING_NAME VARCHAR2(30),
SETTING_VALUE VARCHAR2(128));
BEGIN
-- Select ESM as the algorithm
INSERT INTO ESM_RADON_SETTINGS
VALUES(DBMS_DATA_MINING.ALGO_NAME,
DBMS_DATA_MINING.ALGO_EXPONENTIAL_SMOOTHING);
-- Set accumulation interval to be minute
INSERT INTO ESM_RADON_SETTINGS
VALUES(DBMS_DATA_MINING.EXSM_INTERVAL,
DBMS_DATA_MINING.EXSM_INTERVAL_MIN);
-- Set prediction step to be 30 minutes (half hour)
INSERT INTO ESM_RADON_SETTINGS
VALUES(DBMS_DATA_MINING.EXSM_PREDICTION_STEP,'30');
-- Set ESM model to be Holt-Winters
INSERT INTO ESM_RADON_SETTINGS
VALUES(DBMS_DATA_MINING.EXSM_MODEL,
DBMS_DATA_MINING.EXSM_HW);
-- Set seasonal cycle to be 4
INSERT INTO ESM_RADON_SETTINGS
VALUES(DBMS_DATA_MINING.EXSM_SEASONALITY,'4');
END;
/


-- Build the Exponential Smotthing (ESM) model
BEGIN
DBMS_DATA_MINING.CREATE_MODEL(MODEL_NAME => 'ESM_RADON_SAMPLE',
MINING_FUNCTION => 'TIME_SERIES',
DATA_TABLE_NAME => 'ESM_RADON_DATA',
CASE_ID_COLUMN_NAME => 'AIR_STATION_TIME',
TARGET_COLUMN_NAME => 'RADON',
SETTINGS_TABLE_NAME => 'ESM_RADON_SETTINGS');
END;
/

%sql


-- output setting table
SELECT SETTING_NAME, SETTING_VALUE
FROM USER_MINING_MODEL_SETTINGS
WHERE MODEL_NAME = UPPER('ESM_RADON_SAMPLE')
ORDER BY SETTING_NAME;

%sql


-- get global diagnostics
SELECT NAME,
NUMERIC_VALUE,
STRING_VALUE
FROM DM$VGESM_RADON_SAMPLE
ORDER BY NAME;

%sql




SELECT * FROM JETT.DM$VPESM_RADON_SAMPLE;