将 DataX 执行结果通过钉钉上报

public interface Hook {
    /**     * 返回名字     *     * @return     */    public String getName();
    /**     * TODO 文档     *     * @param jobConf     * @param msg     */    public void invoke(Configuration jobConf, Map<String, Number> msg);
}

/** * Created by jingxing on 14-8-24. * <p/> * job实例运行在jobContainer容器中，它是所有任务的master，负责初始化、拆分、调度、运行、回收、监控和汇报 * 但它并不做实际的数据同步操作 */public class JobContainer extends AbstractContainer {   /**     * jobContainer主要负责的工作全部在start()里面，包括init、prepare、split、scheduler、     * post以及destroy和statistics     */    @Override    public void start() {        LOG.info("DataX jobContainer starts job.");
        boolean hasException = false;        boolean isDryRun = false;        try {            this.startTimeStamp = System.currentTimeMillis();            isDryRun = configuration.getBool(CoreConstant.DATAX_JOB_SETTING_DRYRUN, false);            if(isDryRun) {                LOG.info("jobContainer starts to do preCheck ...");                this.preCheck();            } else {                userConf = configuration.clone();                LOG.debug("jobContainer starts to do preHandle ...");                this.preHandle();
                LOG.debug("jobContainer starts to do init ...");                this.init();                LOG.info("jobContainer starts to do prepare ...");                this.prepare();                LOG.info("jobContainer starts to do split ...");                this.totalStage = this.split();                LOG.info("jobContainer starts to do schedule ...");                this.schedule();                LOG.debug("jobContainer starts to do post ...");                this.post();
                LOG.debug("jobContainer starts to do postHandle ...");                this.postHandle();                LOG.info("DataX jobId [{}] completed successfully.", this.jobId);								// ⚠️ 执行hook ⚠️                this.invokeHooks();            }        }      // 省略    }      /**     * 调用外部hook     */    private void invokeHooks() {        Communication comm = super.getContainerCommunicator().collect();        HookInvoker invoker = new HookInvoker(CoreConstant.DATAX_HOME + "/hook", configuration, comm.getCounter());        invoker.invokeAll();    }  }

public class DingTalkReport implements Hook {	private static final Logger LOG = LoggerFactory.getLogger(DingTalkReport.class);	private static final SimpleDateFormat dateFormat = new SimpleDateFormat(		"yyyy-MM-dd HH:mm:ss");
	@Override	public String getName() {		return "DingTalkReportHook";	}
	@Override	public void invoke(Configuration configuration, Map<String, Number> map) {		LOG.debug(configuration.beautify());		//{writeSucceedRecords=1248,		// readSucceedRecords=1247,		// totalErrorBytes=0,		// writeSucceedBytes=81477,		// byteSpeed=0,		// totalErrorRecords=0,		// recordSpeed=0,		// waitReaderTime=308600221,		// writeReceivedBytes=81477,		// stage=1,		// waitWriterTime=6348796,		// percentage=1.0,		// totalReadRecords=1247,		// writeReceivedRecords=1248,		// readSucceedBytes=81477,		// totalReadBytes=81477}		LOG.debug(map.toString());
		// 从job的json配置中读取自定义的参数		String accessToken = getString(configuration.get("job.dingTalkReporter.accessToken"));		String title = getString(configuration.get("job.dingTalkReporter.title"));		String secret = getString(configuration.get("job.dingTalkReporter.secret"));		String defaultTemplate = "# %s \n > %s \n - 成功读:%d \n - 成功写:%d \n - 等待%d秒";		String time = dateFormat.format(System.currentTimeMillis());		// 好像没办法拿到执行时间，如果需要更多map中没有的信息，可以改写Hook接口		long totalWaitTime = TimeUnit.SECONDS.convert(map.get("waitReaderTime").longValue() + map.get("waitWriterTime").longValue(), TimeUnit.NANOSECONDS);
		String content = String.format(defaultTemplate, title, time, map.get("readSucceedRecords").longValue(), map.get("writeSucceedRecords").longValue(), totalWaitTime);
		DingTalkUtil.send(title, content, accessToken, secret);	}
	String getString(Object obj){		return null == obj ? "" : String.valueOf(obj);	}
}

com.bz.datax.hook.DingTalkReport

{    "job": {        "dingTalkReporter": {            "accessToken": "d9e5b4f89xxxxxxxxxxx",            "secret": "SECc008578xxxxxxx",            "title": "Test表同步"        },        "setting": {            "speed": {                "byte":10485760            },            "errorLimit": {                "record": 0,                "percentage": 0.02            }        },……}

2021-06-03 10:39:08.673 [job-0] INFO  HookInvoker - Invoke hook [DingTalkReportHook], path: /datax/hook/dingtalk2021-06-03 10:39:09.415 [job-0] INFO  DingTalkUtil - Send DingTalk Message Result:ok-0