@Test public void testSourceLineage() { String dataset = Path.getPathWithoutSchemeAndAuthority(sourceDir).toString(); SourceState sourceState = new SourceState(); sourceState.setBroker(jobBroker); initState(sourceState); AvroFileSource fileSource = new AvroFileSource(); List<WorkUnit> workUnits = fileSource.getWorkunits(sourceState); DatasetDescriptor datasetDescriptor = new DatasetDescriptor("hdfs", dataset); for (WorkUnit workUnit : workUnits) { Assert.assertEquals(workUnit.getProp(SOURCE_LINEAGE_KEY), Descriptor.toJson(datasetDescriptor)); } sourceState.setProp(ConfigurationKeys.SOURCE_FILEBASED_PLATFORM, DatasetConstants.PLATFORM_FILE); DatePartitionedJsonFileSource partitionedFileSource = new DatePartitionedJsonFileSource(); workUnits = partitionedFileSource.getWorkunits(sourceState); datasetDescriptor = new DatasetDescriptor("file", dataset); Set<String> partitions = Sets.newHashSet("2017-12", "2018-01"); for (WorkUnit workUnit : workUnits) { if (workUnit instanceof MultiWorkUnit) { DatasetDescriptor finalDatasetDescriptor = datasetDescriptor; ((MultiWorkUnit) workUnit).getWorkUnits().forEach( wu -> verifyPartitionSourceLineage(wu, partitions, finalDatasetDescriptor)); } else { verifyPartitionSourceLineage(workUnit, partitions, datasetDescriptor); } } }
@Override public List<WorkUnit> getWorkunits(SourceState state) { initLogger(state); lineageInfo = LineageInfo.getLineageInfo(state.getBroker()); try { initFileSystemHelper(state); } catch (FileBasedHelperException e) { Throwables.propagate(e); } log.info("Getting work units"); String nameSpaceName = state.getProp(ConfigurationKeys.EXTRACT_NAMESPACE_NAME_KEY); String entityName = state.getProp(ConfigurationKeys.SOURCE_ENTITY); String extractTableName = state.getProp(ConfigurationKeys.EXTRACT_TABLE_NAME_KEY); if (Strings.isNullOrEmpty(extractTableName)) { extractTableName = entityName; } TableType tableType = TableType.valueOf(state.getProp(ConfigurationKeys.EXTRACT_TABLE_TYPE_KEY).toUpperCase()); List<WorkUnitState> previousWorkunits = Lists.newArrayList(state.getPreviousWorkUnitStates()); Set<String> prevFsSnapshot = Sets.newHashSet(); if (!previousWorkunits.isEmpty()) { if (previousWorkunits.get(0).getWorkunit().contains(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)) { prevFsSnapshot.addAll(previousWorkunits.get(0).getWorkunit().getPropAsSet(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)); } else if (state.getPropAsBoolean(ConfigurationKeys.SOURCE_FILEBASED_FS_PRIOR_SNAPSHOT_REQUIRED, ConfigurationKeys.DEFAULT_SOURCE_FILEBASED_FS_PRIOR_SNAPSHOT_REQUIRED)) { throw new RuntimeException(String.format("No '%s' found on state of prior job", ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)); } } List<WorkUnit> workUnits = Lists.newArrayList(); List<WorkUnit> previousWorkUnitsForRetry = this.getPreviousWorkUnitsForRetry(state); log.info("Total number of work units from the previous failed runs: " + previousWorkUnitsForRetry.size()); for (WorkUnit previousWorkUnitForRetry : previousWorkUnitsForRetry) { prevFsSnapshot.addAll(previousWorkUnitForRetry.getPropAsSet(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL)); workUnits.add(previousWorkUnitForRetry); } List<String> currentFsSnapshot = this.getcurrentFsSnapshot(state); List<String> effectiveSnapshot = Lists.newArrayList(); List<String> filesToPull = Lists.newArrayList(); int maxFilesToPull = state.getPropAsInt(ConfigurationKeys.SOURCE_FILEBASED_MAX_FILES_PER_RUN, Integer.MAX_VALUE); int filesSelectedForPull = 0; if (currentFsSnapshot.size() > maxFilesToPull) { Collections.sort(currentFsSnapshot); } for (String file: currentFsSnapshot) { if (prevFsSnapshot.contains(file)) { effectiveSnapshot.add(file); } else if ((filesSelectedForPull++) < maxFilesToPull) { filesToPull.add(file.split(this.splitPattern)[0]); effectiveSnapshot.add(file); } else { } } for (WorkUnit workUnit : previousWorkUnitsForRetry) { workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT, StringUtils.join(effectiveSnapshot, ",")); } if (!filesToPull.isEmpty()) { logFilesToPull(filesToPull); int numPartitions = state.contains(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) && state.getPropAsInt(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) <= filesToPull.size() ? state.getPropAsInt(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) : filesToPull.size(); if (numPartitions <= 0) { throw new IllegalArgumentException("The number of partitions should be positive"); } int filesPerPartition = filesToPull.size() % numPartitions == 0 ? filesToPull.size() / numPartitions : filesToPull.size() / numPartitions + 1; for (int fileOffset = 0; fileOffset < filesToPull.size(); fileOffset += filesPerPartition) { SourceState extractState = new SourceState(); extractState.setProp(ConfigurationKeys.EXTRACT_ID_TIME_ZONE, state.getProp(ConfigurationKeys.EXTRACT_ID_TIME_ZONE, ConfigurationKeys.DEFAULT_EXTRACT_ID_TIME_ZONE)); extractState.setProp(ConfigurationKeys.EXTRACT_IS_FULL_KEY, state.getProp(ConfigurationKeys.EXTRACT_IS_FULL_KEY, ConfigurationKeys.DEFAULT_EXTRACT_IS_FULL)); Extract extract = new Extract(extractState, tableType, nameSpaceName, extractTableName); WorkUnit workUnit = WorkUnit.create(extract); workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT, StringUtils.join(effectiveSnapshot, ",")); List<String> partitionFilesToPull = filesToPull.subList(fileOffset, fileOffset + filesPerPartition > filesToPull.size() ? filesToPull.size() : fileOffset + filesPerPartition); workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL, StringUtils.join(partitionFilesToPull, ",")); if (state.getPropAsBoolean(ConfigurationKeys.SOURCE_FILEBASED_PRESERVE_FILE_NAME, false)) { if (partitionFilesToPull.size() != 1) { throw new RuntimeException("Cannot preserve the file name if a workunit is given multiple files"); } workUnit.setProp(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, workUnit.getProp(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL)); } workUnits.add(workUnit); } log.info("Total number of work units for the current run: " + (workUnits.size() - previousWorkUnitsForRetry.size())); } addLineageSourceInfo(workUnits, state); return workUnits; }
FileBasedSource extends AbstractSource<S, D> { @Override public List<WorkUnit> getWorkunits(SourceState state) { initLogger(state); lineageInfo = LineageInfo.getLineageInfo(state.getBroker()); try { initFileSystemHelper(state); } catch (FileBasedHelperException e) { Throwables.propagate(e); } log.info("Getting work units"); String nameSpaceName = state.getProp(ConfigurationKeys.EXTRACT_NAMESPACE_NAME_KEY); String entityName = state.getProp(ConfigurationKeys.SOURCE_ENTITY); String extractTableName = state.getProp(ConfigurationKeys.EXTRACT_TABLE_NAME_KEY); if (Strings.isNullOrEmpty(extractTableName)) { extractTableName = entityName; } TableType tableType = TableType.valueOf(state.getProp(ConfigurationKeys.EXTRACT_TABLE_TYPE_KEY).toUpperCase()); List<WorkUnitState> previousWorkunits = Lists.newArrayList(state.getPreviousWorkUnitStates()); Set<String> prevFsSnapshot = Sets.newHashSet(); if (!previousWorkunits.isEmpty()) { if (previousWorkunits.get(0).getWorkunit().contains(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)) { prevFsSnapshot.addAll(previousWorkunits.get(0).getWorkunit().getPropAsSet(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)); } else if (state.getPropAsBoolean(ConfigurationKeys.SOURCE_FILEBASED_FS_PRIOR_SNAPSHOT_REQUIRED, ConfigurationKeys.DEFAULT_SOURCE_FILEBASED_FS_PRIOR_SNAPSHOT_REQUIRED)) { throw new RuntimeException(String.format("No '%s' found on state of prior job", ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)); } } List<WorkUnit> workUnits = Lists.newArrayList(); List<WorkUnit> previousWorkUnitsForRetry = this.getPreviousWorkUnitsForRetry(state); log.info("Total number of work units from the previous failed runs: " + previousWorkUnitsForRetry.size()); for (WorkUnit previousWorkUnitForRetry : previousWorkUnitsForRetry) { prevFsSnapshot.addAll(previousWorkUnitForRetry.getPropAsSet(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL)); workUnits.add(previousWorkUnitForRetry); } List<String> currentFsSnapshot = this.getcurrentFsSnapshot(state); List<String> effectiveSnapshot = Lists.newArrayList(); List<String> filesToPull = Lists.newArrayList(); int maxFilesToPull = state.getPropAsInt(ConfigurationKeys.SOURCE_FILEBASED_MAX_FILES_PER_RUN, Integer.MAX_VALUE); int filesSelectedForPull = 0; if (currentFsSnapshot.size() > maxFilesToPull) { Collections.sort(currentFsSnapshot); } for (String file: currentFsSnapshot) { if (prevFsSnapshot.contains(file)) { effectiveSnapshot.add(file); } else if ((filesSelectedForPull++) < maxFilesToPull) { filesToPull.add(file.split(this.splitPattern)[0]); effectiveSnapshot.add(file); } else { } } for (WorkUnit workUnit : previousWorkUnitsForRetry) { workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT, StringUtils.join(effectiveSnapshot, ",")); } if (!filesToPull.isEmpty()) { logFilesToPull(filesToPull); int numPartitions = state.contains(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) && state.getPropAsInt(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) <= filesToPull.size() ? state.getPropAsInt(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) : filesToPull.size(); if (numPartitions <= 0) { throw new IllegalArgumentException("The number of partitions should be positive"); } int filesPerPartition = filesToPull.size() % numPartitions == 0 ? filesToPull.size() / numPartitions : filesToPull.size() / numPartitions + 1; for (int fileOffset = 0; fileOffset < filesToPull.size(); fileOffset += filesPerPartition) { SourceState extractState = new SourceState(); extractState.setProp(ConfigurationKeys.EXTRACT_ID_TIME_ZONE, state.getProp(ConfigurationKeys.EXTRACT_ID_TIME_ZONE, ConfigurationKeys.DEFAULT_EXTRACT_ID_TIME_ZONE)); extractState.setProp(ConfigurationKeys.EXTRACT_IS_FULL_KEY, state.getProp(ConfigurationKeys.EXTRACT_IS_FULL_KEY, ConfigurationKeys.DEFAULT_EXTRACT_IS_FULL)); Extract extract = new Extract(extractState, tableType, nameSpaceName, extractTableName); WorkUnit workUnit = WorkUnit.create(extract); workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT, StringUtils.join(effectiveSnapshot, ",")); List<String> partitionFilesToPull = filesToPull.subList(fileOffset, fileOffset + filesPerPartition > filesToPull.size() ? filesToPull.size() : fileOffset + filesPerPartition); workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL, StringUtils.join(partitionFilesToPull, ",")); if (state.getPropAsBoolean(ConfigurationKeys.SOURCE_FILEBASED_PRESERVE_FILE_NAME, false)) { if (partitionFilesToPull.size() != 1) { throw new RuntimeException("Cannot preserve the file name if a workunit is given multiple files"); } workUnit.setProp(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, workUnit.getProp(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL)); } workUnits.add(workUnit); } log.info("Total number of work units for the current run: " + (workUnits.size() - previousWorkUnitsForRetry.size())); } addLineageSourceInfo(workUnits, state); return workUnits; } }
FileBasedSource extends AbstractSource<S, D> { @Override public List<WorkUnit> getWorkunits(SourceState state) { initLogger(state); lineageInfo = LineageInfo.getLineageInfo(state.getBroker()); try { initFileSystemHelper(state); } catch (FileBasedHelperException e) { Throwables.propagate(e); } log.info("Getting work units"); String nameSpaceName = state.getProp(ConfigurationKeys.EXTRACT_NAMESPACE_NAME_KEY); String entityName = state.getProp(ConfigurationKeys.SOURCE_ENTITY); String extractTableName = state.getProp(ConfigurationKeys.EXTRACT_TABLE_NAME_KEY); if (Strings.isNullOrEmpty(extractTableName)) { extractTableName = entityName; } TableType tableType = TableType.valueOf(state.getProp(ConfigurationKeys.EXTRACT_TABLE_TYPE_KEY).toUpperCase()); List<WorkUnitState> previousWorkunits = Lists.newArrayList(state.getPreviousWorkUnitStates()); Set<String> prevFsSnapshot = Sets.newHashSet(); if (!previousWorkunits.isEmpty()) { if (previousWorkunits.get(0).getWorkunit().contains(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)) { prevFsSnapshot.addAll(previousWorkunits.get(0).getWorkunit().getPropAsSet(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)); } else if (state.getPropAsBoolean(ConfigurationKeys.SOURCE_FILEBASED_FS_PRIOR_SNAPSHOT_REQUIRED, ConfigurationKeys.DEFAULT_SOURCE_FILEBASED_FS_PRIOR_SNAPSHOT_REQUIRED)) { throw new RuntimeException(String.format("No '%s' found on state of prior job", ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)); } } List<WorkUnit> workUnits = Lists.newArrayList(); List<WorkUnit> previousWorkUnitsForRetry = this.getPreviousWorkUnitsForRetry(state); log.info("Total number of work units from the previous failed runs: " + previousWorkUnitsForRetry.size()); for (WorkUnit previousWorkUnitForRetry : previousWorkUnitsForRetry) { prevFsSnapshot.addAll(previousWorkUnitForRetry.getPropAsSet(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL)); workUnits.add(previousWorkUnitForRetry); } List<String> currentFsSnapshot = this.getcurrentFsSnapshot(state); List<String> effectiveSnapshot = Lists.newArrayList(); List<String> filesToPull = Lists.newArrayList(); int maxFilesToPull = state.getPropAsInt(ConfigurationKeys.SOURCE_FILEBASED_MAX_FILES_PER_RUN, Integer.MAX_VALUE); int filesSelectedForPull = 0; if (currentFsSnapshot.size() > maxFilesToPull) { Collections.sort(currentFsSnapshot); } for (String file: currentFsSnapshot) { if (prevFsSnapshot.contains(file)) { effectiveSnapshot.add(file); } else if ((filesSelectedForPull++) < maxFilesToPull) { filesToPull.add(file.split(this.splitPattern)[0]); effectiveSnapshot.add(file); } else { } } for (WorkUnit workUnit : previousWorkUnitsForRetry) { workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT, StringUtils.join(effectiveSnapshot, ",")); } if (!filesToPull.isEmpty()) { logFilesToPull(filesToPull); int numPartitions = state.contains(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) && state.getPropAsInt(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) <= filesToPull.size() ? state.getPropAsInt(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) : filesToPull.size(); if (numPartitions <= 0) { throw new IllegalArgumentException("The number of partitions should be positive"); } int filesPerPartition = filesToPull.size() % numPartitions == 0 ? filesToPull.size() / numPartitions : filesToPull.size() / numPartitions + 1; for (int fileOffset = 0; fileOffset < filesToPull.size(); fileOffset += filesPerPartition) { SourceState extractState = new SourceState(); extractState.setProp(ConfigurationKeys.EXTRACT_ID_TIME_ZONE, state.getProp(ConfigurationKeys.EXTRACT_ID_TIME_ZONE, ConfigurationKeys.DEFAULT_EXTRACT_ID_TIME_ZONE)); extractState.setProp(ConfigurationKeys.EXTRACT_IS_FULL_KEY, state.getProp(ConfigurationKeys.EXTRACT_IS_FULL_KEY, ConfigurationKeys.DEFAULT_EXTRACT_IS_FULL)); Extract extract = new Extract(extractState, tableType, nameSpaceName, extractTableName); WorkUnit workUnit = WorkUnit.create(extract); workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT, StringUtils.join(effectiveSnapshot, ",")); List<String> partitionFilesToPull = filesToPull.subList(fileOffset, fileOffset + filesPerPartition > filesToPull.size() ? filesToPull.size() : fileOffset + filesPerPartition); workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL, StringUtils.join(partitionFilesToPull, ",")); if (state.getPropAsBoolean(ConfigurationKeys.SOURCE_FILEBASED_PRESERVE_FILE_NAME, false)) { if (partitionFilesToPull.size() != 1) { throw new RuntimeException("Cannot preserve the file name if a workunit is given multiple files"); } workUnit.setProp(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, workUnit.getProp(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL)); } workUnits.add(workUnit); } log.info("Total number of work units for the current run: " + (workUnits.size() - previousWorkUnitsForRetry.size())); } addLineageSourceInfo(workUnits, state); return workUnits; }  }
FileBasedSource extends AbstractSource<S, D> { @Override public List<WorkUnit> getWorkunits(SourceState state) { initLogger(state); lineageInfo = LineageInfo.getLineageInfo(state.getBroker()); try { initFileSystemHelper(state); } catch (FileBasedHelperException e) { Throwables.propagate(e); } log.info("Getting work units"); String nameSpaceName = state.getProp(ConfigurationKeys.EXTRACT_NAMESPACE_NAME_KEY); String entityName = state.getProp(ConfigurationKeys.SOURCE_ENTITY); String extractTableName = state.getProp(ConfigurationKeys.EXTRACT_TABLE_NAME_KEY); if (Strings.isNullOrEmpty(extractTableName)) { extractTableName = entityName; } TableType tableType = TableType.valueOf(state.getProp(ConfigurationKeys.EXTRACT_TABLE_TYPE_KEY).toUpperCase()); List<WorkUnitState> previousWorkunits = Lists.newArrayList(state.getPreviousWorkUnitStates()); Set<String> prevFsSnapshot = Sets.newHashSet(); if (!previousWorkunits.isEmpty()) { if (previousWorkunits.get(0).getWorkunit().contains(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)) { prevFsSnapshot.addAll(previousWorkunits.get(0).getWorkunit().getPropAsSet(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)); } else if (state.getPropAsBoolean(ConfigurationKeys.SOURCE_FILEBASED_FS_PRIOR_SNAPSHOT_REQUIRED, ConfigurationKeys.DEFAULT_SOURCE_FILEBASED_FS_PRIOR_SNAPSHOT_REQUIRED)) { throw new RuntimeException(String.format("No '%s' found on state of prior job", ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)); } } List<WorkUnit> workUnits = Lists.newArrayList(); List<WorkUnit> previousWorkUnitsForRetry = this.getPreviousWorkUnitsForRetry(state); log.info("Total number of work units from the previous failed runs: " + previousWorkUnitsForRetry.size()); for (WorkUnit previousWorkUnitForRetry : previousWorkUnitsForRetry) { prevFsSnapshot.addAll(previousWorkUnitForRetry.getPropAsSet(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL)); workUnits.add(previousWorkUnitForRetry); } List<String> currentFsSnapshot = this.getcurrentFsSnapshot(state); List<String> effectiveSnapshot = Lists.newArrayList(); List<String> filesToPull = Lists.newArrayList(); int maxFilesToPull = state.getPropAsInt(ConfigurationKeys.SOURCE_FILEBASED_MAX_FILES_PER_RUN, Integer.MAX_VALUE); int filesSelectedForPull = 0; if (currentFsSnapshot.size() > maxFilesToPull) { Collections.sort(currentFsSnapshot); } for (String file: currentFsSnapshot) { if (prevFsSnapshot.contains(file)) { effectiveSnapshot.add(file); } else if ((filesSelectedForPull++) < maxFilesToPull) { filesToPull.add(file.split(this.splitPattern)[0]); effectiveSnapshot.add(file); } else { } } for (WorkUnit workUnit : previousWorkUnitsForRetry) { workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT, StringUtils.join(effectiveSnapshot, ",")); } if (!filesToPull.isEmpty()) { logFilesToPull(filesToPull); int numPartitions = state.contains(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) && state.getPropAsInt(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) <= filesToPull.size() ? state.getPropAsInt(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) : filesToPull.size(); if (numPartitions <= 0) { throw new IllegalArgumentException("The number of partitions should be positive"); } int filesPerPartition = filesToPull.size() % numPartitions == 0 ? filesToPull.size() / numPartitions : filesToPull.size() / numPartitions + 1; for (int fileOffset = 0; fileOffset < filesToPull.size(); fileOffset += filesPerPartition) { SourceState extractState = new SourceState(); extractState.setProp(ConfigurationKeys.EXTRACT_ID_TIME_ZONE, state.getProp(ConfigurationKeys.EXTRACT_ID_TIME_ZONE, ConfigurationKeys.DEFAULT_EXTRACT_ID_TIME_ZONE)); extractState.setProp(ConfigurationKeys.EXTRACT_IS_FULL_KEY, state.getProp(ConfigurationKeys.EXTRACT_IS_FULL_KEY, ConfigurationKeys.DEFAULT_EXTRACT_IS_FULL)); Extract extract = new Extract(extractState, tableType, nameSpaceName, extractTableName); WorkUnit workUnit = WorkUnit.create(extract); workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT, StringUtils.join(effectiveSnapshot, ",")); List<String> partitionFilesToPull = filesToPull.subList(fileOffset, fileOffset + filesPerPartition > filesToPull.size() ? filesToPull.size() : fileOffset + filesPerPartition); workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL, StringUtils.join(partitionFilesToPull, ",")); if (state.getPropAsBoolean(ConfigurationKeys.SOURCE_FILEBASED_PRESERVE_FILE_NAME, false)) { if (partitionFilesToPull.size() != 1) { throw new RuntimeException("Cannot preserve the file name if a workunit is given multiple files"); } workUnit.setProp(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, workUnit.getProp(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL)); } workUnits.add(workUnit); } log.info("Total number of work units for the current run: " + (workUnits.size() - previousWorkUnitsForRetry.size())); } addLineageSourceInfo(workUnits, state); return workUnits; }  @Override List<WorkUnit> getWorkunits(SourceState state); List<String> getcurrentFsSnapshot(State state); @Override void shutdown(SourceState state); abstract void initFileSystemHelper(State state); }
FileBasedSource extends AbstractSource<S, D> { @Override public List<WorkUnit> getWorkunits(SourceState state) { initLogger(state); lineageInfo = LineageInfo.getLineageInfo(state.getBroker()); try { initFileSystemHelper(state); } catch (FileBasedHelperException e) { Throwables.propagate(e); } log.info("Getting work units"); String nameSpaceName = state.getProp(ConfigurationKeys.EXTRACT_NAMESPACE_NAME_KEY); String entityName = state.getProp(ConfigurationKeys.SOURCE_ENTITY); String extractTableName = state.getProp(ConfigurationKeys.EXTRACT_TABLE_NAME_KEY); if (Strings.isNullOrEmpty(extractTableName)) { extractTableName = entityName; } TableType tableType = TableType.valueOf(state.getProp(ConfigurationKeys.EXTRACT_TABLE_TYPE_KEY).toUpperCase()); List<WorkUnitState> previousWorkunits = Lists.newArrayList(state.getPreviousWorkUnitStates()); Set<String> prevFsSnapshot = Sets.newHashSet(); if (!previousWorkunits.isEmpty()) { if (previousWorkunits.get(0).getWorkunit().contains(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)) { prevFsSnapshot.addAll(previousWorkunits.get(0).getWorkunit().getPropAsSet(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)); } else if (state.getPropAsBoolean(ConfigurationKeys.SOURCE_FILEBASED_FS_PRIOR_SNAPSHOT_REQUIRED, ConfigurationKeys.DEFAULT_SOURCE_FILEBASED_FS_PRIOR_SNAPSHOT_REQUIRED)) { throw new RuntimeException(String.format("No '%s' found on state of prior job", ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT)); } } List<WorkUnit> workUnits = Lists.newArrayList(); List<WorkUnit> previousWorkUnitsForRetry = this.getPreviousWorkUnitsForRetry(state); log.info("Total number of work units from the previous failed runs: " + previousWorkUnitsForRetry.size()); for (WorkUnit previousWorkUnitForRetry : previousWorkUnitsForRetry) { prevFsSnapshot.addAll(previousWorkUnitForRetry.getPropAsSet(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL)); workUnits.add(previousWorkUnitForRetry); } List<String> currentFsSnapshot = this.getcurrentFsSnapshot(state); List<String> effectiveSnapshot = Lists.newArrayList(); List<String> filesToPull = Lists.newArrayList(); int maxFilesToPull = state.getPropAsInt(ConfigurationKeys.SOURCE_FILEBASED_MAX_FILES_PER_RUN, Integer.MAX_VALUE); int filesSelectedForPull = 0; if (currentFsSnapshot.size() > maxFilesToPull) { Collections.sort(currentFsSnapshot); } for (String file: currentFsSnapshot) { if (prevFsSnapshot.contains(file)) { effectiveSnapshot.add(file); } else if ((filesSelectedForPull++) < maxFilesToPull) { filesToPull.add(file.split(this.splitPattern)[0]); effectiveSnapshot.add(file); } else { } } for (WorkUnit workUnit : previousWorkUnitsForRetry) { workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT, StringUtils.join(effectiveSnapshot, ",")); } if (!filesToPull.isEmpty()) { logFilesToPull(filesToPull); int numPartitions = state.contains(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) && state.getPropAsInt(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) <= filesToPull.size() ? state.getPropAsInt(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS) : filesToPull.size(); if (numPartitions <= 0) { throw new IllegalArgumentException("The number of partitions should be positive"); } int filesPerPartition = filesToPull.size() % numPartitions == 0 ? filesToPull.size() / numPartitions : filesToPull.size() / numPartitions + 1; for (int fileOffset = 0; fileOffset < filesToPull.size(); fileOffset += filesPerPartition) { SourceState extractState = new SourceState(); extractState.setProp(ConfigurationKeys.EXTRACT_ID_TIME_ZONE, state.getProp(ConfigurationKeys.EXTRACT_ID_TIME_ZONE, ConfigurationKeys.DEFAULT_EXTRACT_ID_TIME_ZONE)); extractState.setProp(ConfigurationKeys.EXTRACT_IS_FULL_KEY, state.getProp(ConfigurationKeys.EXTRACT_IS_FULL_KEY, ConfigurationKeys.DEFAULT_EXTRACT_IS_FULL)); Extract extract = new Extract(extractState, tableType, nameSpaceName, extractTableName); WorkUnit workUnit = WorkUnit.create(extract); workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FS_SNAPSHOT, StringUtils.join(effectiveSnapshot, ",")); List<String> partitionFilesToPull = filesToPull.subList(fileOffset, fileOffset + filesPerPartition > filesToPull.size() ? filesToPull.size() : fileOffset + filesPerPartition); workUnit.setProp(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL, StringUtils.join(partitionFilesToPull, ",")); if (state.getPropAsBoolean(ConfigurationKeys.SOURCE_FILEBASED_PRESERVE_FILE_NAME, false)) { if (partitionFilesToPull.size() != 1) { throw new RuntimeException("Cannot preserve the file name if a workunit is given multiple files"); } workUnit.setProp(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, workUnit.getProp(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL)); } workUnits.add(workUnit); } log.info("Total number of work units for the current run: " + (workUnits.size() - previousWorkUnitsForRetry.size())); } addLineageSourceInfo(workUnits, state); return workUnits; }  @Override List<WorkUnit> getWorkunits(SourceState state); List<String> getcurrentFsSnapshot(State state); @Override void shutdown(SourceState state); abstract void initFileSystemHelper(State state);  }