获取flink job运行状态脚本

Posted 2023-05-05

参考技术A 需要在主机下安装jq,jq是Linux下面把文本字符串格式化成json格式的工具

#! /bin/bash

current=`date "+%Y-%m-%d %H:%M:%S"`

timeStamp=`date -d "$current" +%s`

currentTimeStamp=$((timeStamp*1000+10#`date "+%N"`/1000000))

echo $currentTimeStamp

flink run -c detect.ZcDataAudit detect_flink-1.0-SNAPSHOT.jar -dbconnect "/mdata1/hadoop/resources/dbconnect.xml" -FlowID 9900001 -taskId 3 -Parallel 36 -billingCycleId 12009 -tms $currentTimeStamp

RESULT=`curl http://xxx.xxx.xxx.xxx:8081/jobs/overview`

state=''

items=`echo $RESULT | jq '.[] | length'`

echo $items

for(( i=0;i<$items;i++ ))

do

str=`echo $RESULT | jq -r .jobs[$i].name`

if [ "$str" = "9900001_ZcDataAudit_$currentTimeStamp" ]

then

state=`echo $RESULT | jq -r .jobs[$i].state`

echo $state

if [ "$state" = "FINISHED" ]

then

echo "任务执行完成"

exit 0

fi

echo "任务执行异常"

exit 1

fi

done

exit 2

flink读取mongodb数据源批处理全量获取实践

本章记录 flink读取mongodb数据源的实践，主要用于 全量 批处理获取mongodb的数据。

我们在前面的文章中已经学习了如何写一个简单的flink-job项目打包jar包运行

详情参考

编写flink任务jar包–helloworld

我们在flink-job项目基础上新增读取mongodb数据源的功能。

引入jar包

使用官方driver连接器

            <dependency>
                <groupId>org.mongodb</groupId>
                <artifactId>mongodb-driver-sync</artifactId>
                <version>4.5.0</version>
            </dependency>

完整pom.xml如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.