需求描述

1， 使用nifi每天跑一次，把confluence的昨天的誰寫了什么題目的記錄同步到一張新表
2， 使用superset設(shè)置一個dashboard，觀看最近兩周的每人每天的貢獻度

相關(guān)數(shù)據(jù)

confluence.CONTENT

CREATE TABLE `CONTENT` (
`CONTENTID`  bigint(20) NOT NULL ,
`HIBERNATEVERSION`  int(11) NOT NULL DEFAULT 0 ,
`CONTENTTYPE`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL ,
`TITLE`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL ,
`LOWERTITLE`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL ,
`VERSION`  int(11) NULL DEFAULT NULL ,
`CREATOR`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL ,
`CREATIONDATE`  datetime NULL DEFAULT NULL ,
`LASTMODIFIER`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL ,
`LASTMODDATE`  datetime NULL DEFAULT NULL ,
`VERSIONCOMMENT`  mediumtext CHARACTER SET utf8 COLLATE utf8_bin NULL ,
`PREVVER`  bigint(20) NULL DEFAULT NULL ,
`CONTENT_STATUS`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL ,
`PAGEID`  bigint(20) NULL DEFAULT NULL ,
`SPACEID`  bigint(20) NULL DEFAULT NULL ,
`CHILD_POSITION`  int(11) NULL DEFAULT NULL ,
`PARENTID`  bigint(20) NULL DEFAULT NULL ,
`MESSAGEID`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL ,
`PLUGINKEY`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL ,
`PLUGINVER`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL ,
`PARENTCCID`  bigint(20) NULL DEFAULT NULL ,
`DRAFTPAGEID`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL ,
`DRAFTSPACEKEY`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL ,
`DRAFTTYPE`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL ,
`DRAFTPAGEVERSION`  int(11) NULL DEFAULT NULL ,
`PARENTCOMMENTID`  bigint(20) NULL DEFAULT NULL ,
`USERNAME`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL ,
PRIMARY KEY (`CONTENTID`)
)
ENGINE=InnoDB
DEFAULT CHARACTER SET=utf8 COLLATE=utf8_bin
ROW_FORMAT=DYNAMIC
;

confluence.user_mapping

CREATE TABLE `user_mapping` (
`user_key`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL ,
`username`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL ,
`lower_username`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL ,
PRIMARY KEY (`user_key`),
UNIQUE INDEX `unq_lwr_username` (`lower_username`) USING BTREE 
)
ENGINE=InnoDB
DEFAULT CHARACTER SET=utf8 COLLATE=utf8_bin
ROW_FORMAT=DYNAMIC
;

nifi_db.Commitments

CREATE TABLE `Commitments` (
`ContentId`  bigint(20) NOT NULL COMMENT '內(nèi)容ID' ,
`WeekOfYear`  varchar(255) CHARACTER SET latin1 COLLATE latin1_swedish_ci NULL DEFAULT NULL COMMENT '年份周數(shù)' ,
`Title`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '內(nèi)容抬頭' ,
`Modifier`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '更新人' ,
`LastModDate`  datetime NOT NULL COMMENT '最后更新時間' ,
`Creator`  varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '創(chuàng)建人' ,
`CreateDate`  datetime NULL DEFAULT NULL COMMENT '創(chuàng)建時間' ,
PRIMARY KEY (`ContentId`, `LastModDate`)
)
ENGINE=InnoDB
DEFAULT CHARACTER SET=latin1 COLLATE=latin1_swedish_ci
ROW_FORMAT=DYNAMIC
;

配置服務(wù)控制器（Controller Service）

DBCPForConfluence(DBCPConnectionPool)

PROPERTIES

DBCPForNiFi_db(DBCPConnectionPool)

PROPERTIES

全量導(dǎo)入歷史貢獻記錄

獲取數(shù)據(jù)（ExecuteSQL）

SCHEDULING

PROPERTIES

SELECT
    a.CONTENTID,
    a.TITLE,
    b.USERNAME AS CREATOR,
    c.USERNAME AS MODIFIER,
    a.CREATIONDATE AS CREATEDATE,
    a.LASTMODDATE,
    YEARWEEK(a.LASTMODDATE) AS WeekOfYear
FROM
    CONTENT a
LEFT JOIN user_mapping b ON a.CREATOR = b.user_key
LEFT JOIN user_mapping c ON a.LASTMODIFIER = c.user_key
WHERE
    a.CONTENTTYPE = 'PAGE'
AND a.spaceid = 98306
AND a.title IS NOT NULL
AND a.PARENTID IS NOT NULL
ORDER BY
    WeekOfYear DESC,
    modifier DESC,
    LASTMODDATE;

格式轉(zhuǎn)化（ConvertAvroToJSON）

直接新增該Processor，默認(rèn)配置即可。

SQL生成（ConvertJSONToSQL）

PROPERTIES

SQL寫入（PutSQL）

PROPERTIES

SQL Statement

注：

該參數(shù)可為空，

當(dāng)為空時，則默認(rèn)執(zhí)行ConvertJSONToSQL處理器提供的SQL。

當(dāng)該參數(shù)不為空時，則忽略ConvertJSONToSQL處理器提供的SQL，只取其數(shù)據(jù)。

本需求場景下，此處建議置空；

REPLACE INTO Commitments (
    ContentId,
    Title,
    Creator,
    Modifier,
    CreateDate,
    LastModDate,
    WeekOfYear
)
VALUES
    (?, ?, ?, ?, ?, ?, ?)

整體流程如圖：

定期（每天）導(dǎo)入貢獻記錄

獲取數(shù)據(jù)（ExecuteSQL）

SCHEDULING

PROPERTIES

SELECT
    a.CONTENTID,
    a.TITLE,
    b.USERNAME AS CREATOR,
    c.USERNAME AS MODIFIER,
    a.CREATIONDATE AS CREATEDATE,
    a.LASTMODDATE,
    YEARWEEK(a.LASTMODDATE) AS WeekOfYear
FROM
    CONTENT a
LEFT JOIN user_mapping b ON a.CREATOR = b.user_key
LEFT JOIN user_mapping c ON a.LASTMODIFIER = c.user_key
WHERE
    a.CONTENTTYPE = 'PAGE'
AND a.spaceid = 98306
AND a.title IS NOT NULL
AND a.PARENTID IS NOT NULL
AND WEEK (a.CREATIONDATE) = WEEK (CURRENT_DATE())
ORDER BY
    WeekOfYear DESC,
    modifier DESC,
    LASTMODDATE;

后續(xù)處理器配置同全量導(dǎo)入即可（見上文）；

整體流程如下圖：