先通过pipeLineId和DataMediaPairId在数据库里查找对应的tableStat，如果有，则增量更新对应的Table统计状态， 如果没有则将该数据插入
508149
public void updateTableStat(TableStat stat) {
        Assert.assertNotNull(stat);
        int affect = tableStatDao.modifyTableStat(tableStatModelToDo(stat));
        if (affect == 0) {
            tableStatDao.insertTableStat(tableStatModelToDo(stat));
        }

        if (stat.getStartTime() != null && stat.getEndTime() != null) {
            if (statUnit <= 0) {
                insertBehaviorHistory(stat);
            } else {
                synchronized (tableStats) {
                    // 插入历史数据表
                    TableStat old = tableStats.get(stat.getDataMediaPairId());
                    if (old != null) {
                        //合并数据
                        old.setInsertCount(stat.getInsertCount() + old.getInsertCount());
                        old.setUpdateCount(stat.getUpdateCount() + old.getUpdateCount());
                        old.setDeleteCount(stat.getDeleteCount() + old.getDeleteCount());
                        old.setFileCount(stat.getFileCount() + old.getFileCount());
                        old.setFileSize(stat.getFileSize() + old.getFileSize());
                        if (stat.getEndTime().after(old.getEndTime())) {
                            old.setEndTime(stat.getEndTime());
                        }
                        if (stat.getStartTime().before(old.getStartTime())) {
                            old.setStartTime(stat.getStartTime());
                        }
                    } else {
                        tableStats.put(stat.getDataMediaPairId(), stat);
                    }
                }
            }
        }
    }
62661
public void doBatchAdd(@FormGroup("batchDataMediaPairInfo") Group batchDataMediaPairInfo,
                           @Param("pipelineId") Long pipelineId,
                           @FormField(name = "formBatchDataMediaPairError", group = "batchDataMediaPairInfo") CustomErrors err,
                           Navigator nav) throws Exception {
        String batchPairContent = batchDataMediaPairInfo.getField("batchPairContent").getStringValue();
        List<String> StringPairs = Arrays.asList(batchPairContent.split("\r\n"));
        try {
            for (String stringPair : StringPairs) {
                List<String> pairData = Arrays.asList(stringPair.split(","));
                if (pairData.size() < 4) {
                    throw new ManagerException("[" + stringPair + "] the line not all parameters");
                }
                // build the pair source
                DataMedia sourceDataMedia = new DataMedia();
                DataMediaSource sourceDataMediaSource = dataMediaSourceService.findById(Long.parseLong(StringUtils.trimToNull(pairData.get(2))));
                sourceDataMedia.setNamespace(StringUtils.trimToNull(pairData.get(0)));
                sourceDataMedia.setName(StringUtils.trimToNull(pairData.get(1)));
                sourceDataMedia.setSource(sourceDataMediaSource);
                Long sourceMediaId = dataMediaService.createReturnId(sourceDataMedia);
                sourceDataMedia.setId(sourceMediaId);
                // build the pair target
                DataMedia targetDataMedia = new DataMedia();
                Long weight = 5L;
                if (StringUtils.isNumeric(pairData.get(3)) && pairData.size() <= 5) {// 如果是纯数字，那说明是简化配置模式
                    DataMediaSource targetDataMediaSource = dataMediaSourceService.findById(Long.parseLong(StringUtils.trimToNull(pairData.get(3))));
                    targetDataMedia.setNamespace(StringUtils.trimToNull(pairData.get(0)));
                    targetDataMedia.setName(StringUtils.trimToNull(pairData.get(1)));
                    targetDataMedia.setSource(targetDataMediaSource);
                    Long targetMediaId = dataMediaService.createReturnId(targetDataMedia);
                    targetDataMedia.setId(targetMediaId);

                    if (pairData.size() >= 5) {
                        weight = Long.parseLong(StringUtils.trimToNull(pairData.get(4)));
                    }
                } else {
                    DataMediaSource targetDataMediaSource = dataMediaSourceService.findById(Long.parseLong(StringUtils.trimToNull(pairData.get(5))));
                    targetDataMedia.setNamespace(StringUtils.trimToNull(pairData.get(3)));
                    targetDataMedia.setName(StringUtils.trimToNull(pairData.get(4)));
                    targetDataMedia.setSource(targetDataMediaSource);
                    Long targetMediaId = dataMediaService.createReturnId(targetDataMedia);
                    targetDataMedia.setId(targetMediaId);

                    if (pairData.size() >= 7) {
                        weight = Long.parseLong(StringUtils.trimToNull(pairData.get(6)));
                    }
                }

                // build the pair
                DataMediaPair dataMediaPair = new DataMediaPair();
                dataMediaPair.setSource(sourceDataMedia);
                dataMediaPair.setTarget(targetDataMedia);
                dataMediaPair.setPushWeight(weight);
                dataMediaPair.setPipelineId(pipelineId);

                dataMediaPairService.createIfNotExist(dataMediaPair);
            }
        } catch (Exception e) {
            err.setMessage("invalidBatchDataMediaPair");
            return;
        }
        nav.redirectToLocation("dataMediaPairList.htm?pipelineId=" + pipelineId);
    }
355793
public void get_table( String sql, String args[], TableResult tbl ) throws jsqlite.Exception {
        tbl.clear();
        if (!is3()) {
            try {
                exec(sql, tbl, args);
            } catch (jsqlite.Exception e) {
                if (tbl.maxrows <= 0 || !tbl.atmaxrows) {
                    throw e;
                }
            }
        } else {
            synchronized (this) {
                /* only one statement !!! */
                Vm vm = compile(sql, args);
                if (tbl.maxrows > 0) {
                    while( tbl.nrows < tbl.maxrows && vm.step(tbl) ) {
                        set_last_error(vm.error_code);
                    }
                } else {
                    while( vm.step(tbl) ) {
                        set_last_error(vm.error_code);
                    }
                }
                vm.finalize();
            }
        }
    }
190025
function () {
      var rowCounts = new Uint32Array(nrows);
      var columnCounts = new Uint32Array(ncols);
      for (var i = 0; i < nrows; i++) {
        var obj = matrix[i];
        if (obj.values != null) {
          // trim to size
          var values = new Uint16Array(obj.length);
          var indices = new Uint32Array(obj.length);
          values.set(obj.values.slice(0, obj.length));
          indices.set(obj.indices.slice(0, obj.length));
          matrix[i] = {values: values, indices: indices};
          rowCounts[i] = values.length;
          for (var j = 0; j < obj.length; j++) {
            columnCounts[indices[j]] += 1;
          }
        }
      }
      dataset = new morpheus.Dataset({
        rows: nrows, columns: ncols, name: morpheus.Util.getBaseFileName(morpheus.Util
          .getFileName(fileOrUrl)), array: matrix, dataType: dataType, defaultValue: 0
      });

      dataset.getRowMetadata().add('count>0').array = rowCounts;
      dataset.getColumnMetadata().add('count>0').array = columnCounts;
    }
609562
@Override
	public void updateJobStatus(long instanceId, JobStatus jobStatus) {
		logger.entering(CLASSNAME, "updateJobStatus", new Object[] {instanceId, jobStatus});
		if (logger.isLoggable(Level.FINE)) {
			logger.fine("Updating Job Status to: " + jobStatus.getBatchStatus());
		}
		Connection conn = null;
		PreparedStatement statement = null;
		try {
			conn = getConnection();
			statement = conn.prepareStatement("UPDATE jobstatus SET obj = ? WHERE id = ?");
			statement.setBytes(1, serializeObject(jobStatus));
			statement.setLong(2, instanceId);
			statement.executeUpdate();
		} catch (SQLException e) {
			throw new PersistenceException(e);
		} catch (IOException e) {
			throw new PersistenceException(e);
		} finally {
			cleanupConnection(conn, null, statement);
		}
		logger.exiting(CLASSNAME, "updateJobStatus");
	}
1370150
Delta newUnpublishedDatabusEventUpdate(String tableName, String updateType, String datetime) {
        return Deltas.mapBuilder()
                .update("tables", Deltas.setBuilder().add(ImmutableMap.of("table", tableName, "date", datetime, "event", updateType)).build())
                .build();
    }
744534
private void recovery(Long pipelineId) {
        List<Long> currentProcessIds = stageMonitor.getCurrentProcessIds(false);
        for (Long processId : currentProcessIds) {
            String path = StagePathUtils.getProcess(pipelineId, processId);
            try {
                byte[] bytes = zookeeper.readData(path);
                ProcessNodeEventData nodeData = JsonUtils.unmarshalFromByte(bytes, ProcessNodeEventData.class);
                if (nodeData.getStatus().isUnUsed()) {// 加入未使用的processId
                    addReply(processId);
                }
            } catch (ZkException e) {
                logger.error("SelectStageListener", e);
            }
        }
    }
418066
protected void handleShowTableName(EntityDesc entityDesc, TableMeta tableMeta) {
    if (showTableName || isNameDifferentBetweenEntityAndTable(entityDesc)) {
      entityDesc.setShowTableName(true);
    }
  }
1343547
public EtlEventData await(Long pipelineId) throws InterruptedException {
        Assert.notNull(pipelineId);
        PermitMonitor permitMonitor = ArbitrateFactory.getInstance(pipelineId, PermitMonitor.class);
        permitMonitor.waitForPermit();// 阻塞等待授权

        ExtractStageListener extractStageListener = ArbitrateFactory.getInstance(pipelineId, ExtractStageListener.class);
        Long processId = extractStageListener.waitForProcess(); // 符合条件的processId

        ChannelStatus status = permitMonitor.getChannelPermit();
        if (status.isStart()) {// 即时查询一下当前的状态，状态随时可能会变
            // 根据pipelineId+processId构造对应的path
            String path = StagePathUtils.getSelectStage(pipelineId, processId);

            try {
                byte[] data = zookeeper.readData(path);
                EtlEventData eventData = JsonUtils.unmarshalFromByte(data, EtlEventData.class);

                Node node = LoadBalanceFactory.getNextTransformNode(pipelineId);// 获取下一个处理节点信息
                if (node == null) {// 没有后端节点
                    // TerminEventData termin = new TerminEventData();
                    // termin.setPipelineId(pipelineId);
                    // termin.setType(TerminType.ROLLBACK);
                    // termin.setCode("no_node");
                    // termin.setDesc(MessageFormat.format("pipeline[{}] extract stage has no node!", pipelineId));
                    // terminEvent.single(termin);
                    throw new ArbitrateException("Extract_single", "no next node");
                } else {
                    eventData.setNextNid(node.getId());
                    return eventData;// 只有这一条路返回
                }
            } catch (ZkNoNodeException e) {
                logger.error("pipeline[{}] processId[{}] is invalid , retry again", pipelineId, processId);
                return await(pipelineId);// /出现节点不存在，说明出现了error情况,递归调用重新获取一次
            } catch (ZkException e) {
                throw new ArbitrateException("Extract_await", e.getMessage(), e);
            }
        } else {
            logger.warn("pipelineId[{}] extract ignore processId[{}] by status[{}]", new Object[] { pipelineId,
                    processId, status });
                    
            // 释放下processId，因为load是等待processId最小值完成Tranform才继续，如果这里不释放，会一直卡死等待
            String path = StagePathUtils.getProcess(pipelineId, processId);
            zookeeper.delete(path);
            return await(pipelineId);// 递归调用
        }

    }
1153480
function activitiesMonitoringStep(){
	pipeRunStep.call(this);

	this.label = "Monitoring DataWorks activities for completion";

	//public APIs
	this.run = function( callback ){
		var logger = this.pipeRunStats.logger;
		var pipeRunStats = this.pipeRunStats;

		//Get the DataWorks instance
		var dwInstance = pipeRunStats.dwInstance;

		var stepStats = this.stats;
		stepStats.numRunningActivities = 0;
		_.forOwn( pipeRunStats.getTableStats(), function(value, key ){
			stepStats.numRunningActivities++;
		})

		stepStats.numFinishedActivities = 0;

		//convenience method
		var expectedLength = this.getPipeRunner().getSourceTables().length;
		var formatStepMessage = function(){
			var percent = ((stepStats.numFinishedActivities/expectedLength)*100).toFixed(1);
			this.setPercentCompletion( percent );
			var message = stepStats.numFinishedActivities + " DataWorks activities completed (" + percent + "%)";
			this.setStepMessage( message );
		}.bind(this);

		var sendAlert = function(){
			logger.info("Sending alert about possible DataWorks activities hung to metrics-collector");
			async.forEachOf( pipeRunStats.getTableStats(), function(tableStats, tableName, callback ){
				if ( !tableStats.activityDone ){
					var d = moment();
					var props = {
						idsite: "data.pipes",
						type : "pipe_alert",
						activityId: tableStats.activityId,
						activityRunId: tableStats.activityRunId,
						tableName: tableStats.tableName,
						numRecords: tableStats.numRecords,
						date: d.year() + "-" + (d.month() + 1) + "-" + d.date(),
						d: d.format()
					};
					request.get( {url: "http://metrics-collector.mybluemix.net/tracker", qs: props}, function(err, response, body){
						if ( !err ){
							logger.info("Successfully logged an alert to metrics-collector");
						}
						return callback( err );
					});
				}
			},function(err){
				if ( err ){
					logger.error("Unable to create DataWorks activity hung alert: " + err );
				}
			});
		}

		//Start monitoring the activities
		var timeout = moment.duration(5, 'minutes').asMilliseconds();	//timeout to raise possible dataworks hung alerts when no activities has completed.
		var alertSent = false;
		var start = moment();
		formatStepMessage();
		var monitor = function(){
			async.forEachOf( pipeRunStats.getTableStats(), function(tableStats, tableName, callback ){
				if ( !tableStats.activityDone ){
					dwInstance.monitorActivityRun( tableStats.activityId, tableStats.activityRunId, function( err, activityRun ){
						if ( err ){
							return callback(err);
						}
						if ( dwInstance.isFinished( activityRun.status ) ){
							logger.trace("ActivityRun %s complete", tableStats.activityRunId);
							stepStats.numFinishedActivities++;
							stepStats.numRunningActivities--;
							tableStats.activityDone = true;
							formatStepMessage();

							//Reset the timeout start
							start = moment();
						}
						return callback();
					})
				}else{
					return callback();
				}
			},function(err){
				if ( err ){
					return callback(err);
				}
				if ( stepStats.numRunningActivities > 0 ){
					//Schedule another round
					if ( moment.duration( moment().diff(start) ).asMilliseconds() > timeout ){
						if ( !alertSent ){
							alertSent = true;
							sendAlert();
						}
					}
					return setTimeout( monitor, 10000 );
				}
				var message = "DataWorks activities are complete";
				logger.info( message );
				stepStats.status = message;
				this.setStepMessage(message);
				return callback();
			}.bind(this));
		}.bind(this);
		setTimeout( monitor, 10000 );
	}
}
1313877
public static String getDatasetUri(Table table) {
    return HIVE_DATASETS_CONFIG_PREFIX + table.getDbName() + Path.SEPARATOR + table.getTableName();
  }
826132
@Override
    public void updateReport(AllTablesReportDelta delta) {
        checkNotNull(delta, "delta");

        updateMetadata(delta);

        if (delta.getTable().isPresent()) {
            updateTableData(delta, delta.getTable().get());
        }
    }
1371500
@SuppressWarnings("resource")
	public static Table getTableMeta(DataSource ds, String tableName) {
		final Table table = Table.create(tableName);
		Connection conn = null;
		ResultSet rs = null;
		try {
			conn = ds.getConnection();
			final DatabaseMetaData metaData = conn.getMetaData();
			// 获得主键
			rs = metaData.getPrimaryKeys(conn.getCatalog(), null, tableName);
			while (rs.next()) {
				table.addPk(rs.getString("COLUMN_NAME"));
			}

			// 获得列
			rs = metaData.getColumns(conn.getCatalog(), null, tableName, null);
			while (rs.next()) {
				table.setColumn(Column.create(tableName, rs));
			}
		} catch (SQLException e) {
			throw new DbRuntimeException("Get columns error!", e);
		} finally {
			DbUtil.close(rs, conn);
		}

		return table;
	}
972294
public DelayStat findRealtimeDelayStat(Long pipelineId) {
        Assert.assertNotNull(pipelineId);
        DelayStatDO delayStatDO = delayStatDao.findRealtimeDelayStat(pipelineId);
        DelayStat delayStat = new DelayStat();
        if (delayStatDO != null) {
            delayStat = delayStatDOToModel(delayStatDO);
        }
        return delayStat;
    }
1103832
private JSONObject tableRecSyncHelper(byte[] imgData, long timeout, String resultType) {
        JSONObject res = tableRecognitionAsync(imgData, null);
        if (res.has("error_code")) {
            return res;
        }
        String reqId = res.getJSONArray("result").getJSONObject(0).getString("request_id");
        long start = Calendar.getInstance().getTimeInMillis();
        long sleepInterval = 2000;
        JSONObject result;
        while (true) {
            long now = Calendar.getInstance().getTimeInMillis();
            if (now - start > timeout) {
                // 超时
                return AipError.ASYNC_TIMEOUT_ERROR.toJsonResult();
            }
            result = getTableResultHelper(reqId, resultType);
            if (result.has("error_code")) {
                // 返回错误
                return result;
            }
            int retCode = result.getJSONObject("result").getInt("ret_code");
            if (retCode == OcrConsts.ASYNC_TASK_STATUS_FINISHED) {
                return result;
            }
            else {
                try {
                    Thread.sleep(sleepInterval);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
617693
private DataMedia doToModel(DataMediaDO dataMediaDo) {
        DataMedia dataMedia = null;
        try {
            DataMediaSource dataMediaSource = dataMediaSourceService.findById(dataMediaDo.getDataMediaSourceId());
            if (dataMediaSource.getType().isMysql() || dataMediaSource.getType().isOracle()) {
                dataMedia = JsonUtils.unmarshalFromString(dataMediaDo.getProperties(), DbDataMedia.class);
                dataMedia.setSource(dataMediaSource);
            } else if (dataMediaSource.getType().isNapoli() || dataMediaSource.getType().isMq()) {
                dataMedia = JsonUtils.unmarshalFromString(dataMediaDo.getProperties(), MqDataMedia.class);
                dataMedia.setSource(dataMediaSource);
            }

            dataMedia.setId(dataMediaDo.getId());
            dataMedia.setGmtCreate(dataMediaDo.getGmtCreate());
            dataMedia.setGmtModified(dataMediaDo.getGmtModified());

        } catch (Exception e) {
            logger.error("ERROR ## change the dataMedia Do to Model has an exception");
            throw new ManagerException(e);
        }

        return dataMedia;
    }
54315
public static DataMediaPair findDataMediaPair(Pipeline pipeline, Long pairId) {
        Assert.notNull(pipeline);
        for (DataMediaPair pair : pipeline.getPairs()) {
            if (pair.getId().equals(pairId)) {
                return pair;
            }
        }

        throw new ConfigException("no such DataMediaPair , the pairId = " + pairId);
    }