@@ -0,0 +1,3094 @@
+package metadata.etl.lhotse.extractor;%0A%0Aimport metadata.etl.lhotse.LzExecMessage;%0Aimport metadata.etl.lhotse.LzTaskExecRecord;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0Aimport wherehows.common.schemas.LineageRecord;%0Aimport wherehows.common.utils.XmlParser;%0A%0Aimport java.util.ArrayList;%0Aimport java.util.List;%0A%0A/**%0A * Created by lake on 17-6-6.%0A */%0Apublic class Hdfs2HBaseLineageExtractor implements BaseLineageExtractor %7B%0A    private static final Logger LOG = LoggerFactory.getLogger(Hdfs2HBaseLineageExtractor.class);%0A%0A    @Override%0A    public List%3CLineageRecord%3E getLineageRecord(String logLocation, LzExecMessage message, int defaultDatabaseId, String logPath) %7B%0A        List%3CLineageRecord%3E lineageRecords = new ArrayList%3C%3E();%0A        LzTaskExecRecord lzTaskExecRecord = message.lzTaskExecRecord;%0A        if (lzTaskExecRecord.flowId == null) return lineageRecords;%0A%0A%0A        try %7B%0A            LOG.info(%22HBase start to parse the log: %7B%7D%22, logLocation);%0A            XmlParser xmlParser = new XmlParser(logLocation);%0A%0A            String srcFile = xmlParser.getExtProperty(%22extProperties/entry/dataPath%22);%0A            String targetHBaseTable = xmlParser.getExtProperty(%22extProperties/entry/hbase.table%22);%0A%0A            long flowExecId = lzTaskExecRecord.flowId;%0A            String flowPath = String.format(%22%25s:%25s%22, lzTaskExecRecord.projectName, lzTaskExecRecord.workflowName);%0A%0A            long taskId = Long.parseLong(lzTaskExecRecord.taskId);%0A            String taskName = lzTaskExecRecord.taskName;%0A%0A            String operation = %22hdfs file data to hbase table%22;%0A%0A            long num = 0L;%0A%0A            /** src record */%0A            LineageRecord srcRecord = new LineageRecord(lzTaskExecRecord.appId, flowExecId, taskName, taskId);%0A            srcRecord.setDatasetInfo(defaultDatabaseId, srcFile, %22hdfs%22);%0A            srcRecord.setOperationInfo(%22source%22, operation, num, num,%0A                    num, num, lzTaskExecRecord.taskStartTime, lzTaskExecRecord.taskEndTime, flowPath);%0A            srcRecord.setAbstractObjectName(srcFile);%0A            srcRecord.setFullObjectName(srcFile);%0A            LOG.info(%22the source record is: %7B%7D%22, srcRecord.toDatabaseValue());%0A            lineageRecords.add(srcRecord);%0A%0A%0A            /** target record */%0A            LineageRecord targetRecord = new LineageRecord(lzTaskExecRecord.appId, flowExecId, taskName, taskId);%0A            targetRecord.setDatasetInfo(defaultDatabaseId, targetHBaseTable, %22hbase%22);%0A            targetRecord.setOperationInfo(%22target%22, operation, num, num,%0A                    num, num, lzTaskExecRecord.taskStartTime, lzTaskExecRecord.taskEndTime, flowPath);%0A            targetRecord.setAbstractObjectName(%22/%22 + targetHBaseTable);%0A            targetRecord.setFullObjectName(%22/%22 + targetHBaseTable);%0A            LOG.info(%22the source record is: %7B%7D%22, targetRecord.toDatabaseValue());%0A            lineageRecords.add(targetRecord);%0A%0A            return lineageRecords;%0A%0A        %7D catch (Exception e) %7B%0A            e.printStackTrace();%0A            LOG.error(e.getMessage());%0A        %7D%0A%0A        return lineageRecords;%0A    %7D%0A%7D%0A

add hdfs to hbase lineage extractor