import org.apache.spark.Partition; //导入依赖的package包/类
protected void checkUserPartition(JavaRDD<String> userRDD) {
  System.out.println("hhhhh");
  List<Partition> partitios = userRDD.partitions();
  System.out.println(partitios.size());
  int[] partitionIds = new int[partitios.size()];
  for (int i = 0; i < partitios.size(); i++) {
    int index = partitios.get(i).index();
    partitionIds[i] = index;
  }

  List<String>[] userIPs = userRDD.collectPartitions(partitionIds);
  for (int i = 0; i < userIPs.length; i++) {
    List<String> iuser = userIPs[i];
    System.out.println(i + " partition");
    System.out.println(iuser.toString());
  }
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public Partition[] getPartitions() {
  try {
    final List<? extends Source<T>> partitionedSources =
        microbatchSource.split(options.get());
    final Partition[] partitions = new CheckpointableSourcePartition[partitionedSources.size()];
    for (int i = 0; i < partitionedSources.size(); i++) {
      partitions[i] =
          new CheckpointableSourcePartition<>(
              id(), i, partitionedSources.get(i), EmptyCheckpointMark.get());
    }
    return partitions;
  } catch (Exception e) {
    throw new RuntimeException("Failed to create partitions.", e);
  }
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public void initIterator(Partition dp, S config) {

    int id = config.getRddId();

    NewHadoopPartition split = (NewHadoopPartition) dp;

    TaskAttemptID attemptId = DeepSparkHadoopMapReduceUtil
            .newTaskAttemptID(jobTrackerId, id, true, split.index(), 0);

    Configuration configuration = getHadoopConfig(config);

    TaskAttemptContext hadoopAttemptContext = DeepSparkHadoopMapReduceUtil
            .newTaskAttemptContext(configuration, attemptId);

    try {
        reader = inputFormat.createRecordReader(split.serializableHadoopSplit().value(), hadoopAttemptContext);
        reader.initialize(split.serializableHadoopSplit().value(), hadoopAttemptContext);
    } catch (IOException | InterruptedException e) {
        throw new DeepGenericException(e);
    }
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public Partition[] getPartitions(ExtractorConfig<T> config) {

    GetPartitionsAction<T> getPartitionsAction = new GetPartitionsAction<>(config);

    channel.writeAndFlush(getPartitionsAction);

    Response response;
    boolean interrupted = false;
    for (; ; ) {
        try {
            response = answer.take();
            break;
        } catch (InterruptedException ignore) {
            interrupted = true;
        }
    }

    if (interrupted) {
        Thread.currentThread().interrupt();
    }

    return ((GetPartitionsResponse) response).getPartitions();
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public void initIterator(Partition dp, ExtractorConfig<T> config) {
    InitIteratorAction<T> initIteratorAction = new InitIteratorAction<>(dp, config);

    channel.writeAndFlush(initIteratorAction);

    Response response;
    boolean interrupted = false;
    for (; ; ) {
        try {
            response = answer.take();
            break;
        } catch (InterruptedException ignore) {
            interrupted = true;
        }
    }

    if (interrupted) {
        Thread.currentThread().interrupt();
    }
    return;
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public Partition[] getPartitions(S config) {
    jdbcDeepJobConfig = initConfig(config, jdbcDeepJobConfig);

    int upperBound = jdbcDeepJobConfig.getUpperBound();
    int lowerBound = jdbcDeepJobConfig.getLowerBound();
    int numPartitions = jdbcDeepJobConfig.getNumPartitions();
    int length = 1 + upperBound - lowerBound;
    Partition [] result = new Partition[numPartitions];
    for(int i=0; i<numPartitions; i++) {
        int start = lowerBound + lowerBound + ((i * length) / numPartitions);
        int end = lowerBound + (((i + 1) * length) / numPartitions) - 1;
        result[i] = new JdbcPartition(i, start, end);
    }
    return result;
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
/**
 * Initialized the reader
 * 
 * @param p
 *            Spark partition.
 * @throws Exception
 */
public void init(Partition p) throws Exception {
    Class.forName(jdbcDeepJobConfig.getDriverClass());
    conn = DriverManager.getConnection(jdbcDeepJobConfig.getConnectionUrl(),
            jdbcDeepJobConfig.getUsername(),
            jdbcDeepJobConfig.getPassword());
    Statement statement = conn.createStatement();
    SelectQuery query = jdbcDeepJobConfig.getQuery();
    JdbcPartition jdbcPartition = (JdbcPartition)p;
    if(jdbcDeepJobConfig.getNumPartitions() > 1) {
        Column partitionKey = jdbcDeepJobConfig.getPartitionKey();
        query.getWhereClause().addCondition(BinaryCondition.lessThan(partitionKey, jdbcPartition.upper(), true))
                .addCondition(BinaryCondition.greaterThan(partitionKey, jdbcPartition.lower(), true));
    }
    resultSet = statement.executeQuery(query.toString());
    // Fetches first element
    this.hasNext = resultSet.next();
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public PartitionGroup[] coalesce(int maxPartitions, RDD<?> parent) {
    if (maxPartitions != parent.getNumPartitions()) {
        throw new IllegalArgumentException("Cannot use " + getClass().getSimpleName() +
                " with a different number of partitions to the parent RDD.");
    }
    List<Partition> partitions = Arrays.asList(parent.getPartitions());
    PartitionGroup[] groups = new PartitionGroup[partitions.size()];

    for (int i = 0; i < partitions.size(); i++) {
        Seq<String> preferredLocations = parent.getPreferredLocations(partitions.get(i));
        scala.Option<String> preferredLocation = scala.Option.apply
                (preferredLocations.isEmpty() ? null : preferredLocations.apply(0));
        PartitionGroup group = new PartitionGroup(preferredLocation);
        List<Partition> partitionsInGroup =
                partitions.subList(i, maxEndPartitionIndexes.get(i) + 1);
        group.partitions().append(JavaConversions.asScalaBuffer(partitionsInGroup));
        groups[i] = group;
    }
    return groups;
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public Iterator<double[]> compute(final Partition partition, final TaskContext context) {
    ProgrammingError.throwIfNull(partition, context);
    if (partition instanceof Partition2D) {
        return this.compute((Partition2D) partition, context);
    } else {
        throw new IllegalArgumentException();
    }
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public Iterator<MatrixStore<N>> compute(final Partition partition, final TaskContext context) {
    ProgrammingError.throwIfNull(partition, context);
    if (partition instanceof Partition2D) {
        return this.compute((Partition2D) partition, context);
    } else {
        throw new IllegalArgumentException();
    }
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public scala.collection.Iterator<TReturn> compute(Partition split, TaskContext context) {
    String regionEdgesFamilyPath = this.regionsPaths.get(split.index());
    log.info("Running Mizo on region #{} located at: {}", split.index(), regionEdgesFamilyPath);

    return createRegionIterator(createRegionRelationsIterator(regionEdgesFamilyPath));
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public Partition[] getPartitions() {
    return Iterators.toArray(IntStream
            .range(0, this.regionsPaths.size())
            .mapToObj(i -> (Partition) () -> i)
            .iterator(), Partition.class);
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public scala.collection.Iterator<WindowedValue<T>> compute(final Partition split,
                                                           final TaskContext context) {
  final MetricsContainer metricsContainer = metricsAccum.localValue().getContainer(stepName);

  @SuppressWarnings("unchecked")
  final BoundedSource.BoundedReader<T> reader = createReader((SourcePartition<T>) split);

  final Iterator<WindowedValue<T>> readerIterator =
      new ReaderToIteratorAdapter<>(metricsContainer, reader);

  return new InterruptibleIterator<>(context, JavaConversions.asScalaIterator(readerIterator));
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public scala.collection.Iterator<Map.Entry<Key, Value>> compute(final Partition split, final TaskContext context) {
    final ByteArrayInputStream bais = new ByteArrayInputStream(serialisedConfiguration);
    final Configuration configuration = new Configuration();
    try {
        configuration.readFields(new DataInputStream(bais));
        bais.close();
    } catch (final IOException e) {
        throw new RuntimeException("IOException deserialising Configuration from byte array", e);
    }
    return new InterruptibleIterator<>(context,
            JavaConversions.asScalaIterator(new RFileReaderIterator(split, context, configuration, auths)));
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
public RFileReaderIterator(final Partition partition,
                           final TaskContext taskContext,
                           final Configuration configuration,
                           final Set<String> auths) {
    this.partition = partition;
    this.taskContext = taskContext;
    this.configuration = configuration;
    this.auths = auths;
    try {
        init();
    } catch (final IOException e) {
        throw new RuntimeException("IOException initialising RFileReaderIterator", e);
    }
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public Iterator<IndexedContent> compute(Partition split, TaskContext context) {
    try {
        Path path = new Path(partitions[split.index()].getPath());
        java.util.Iterator<Content> iterator = new ContentIterator(path, NutchConfiguration.create(), contentTypeFilter);
        return new IndexedContentIterator(iterator);
    } catch (Exception e) {
        LOG.error(e.getMessage(), e);
        throw new RuntimeException(e);
    }
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public Iterator<Content> compute(Partition split, TaskContext context) {
    try {
        Path path = new Path(partitions[split.index()].getPath());
        return new ContentIterator(path, NutchConfiguration.create(), contentTypeFilter);
    } catch (Exception e) {
        LOG.error(e.getMessage(), e);
        throw new RuntimeException(e);
    }
}
 

import org.apache.spark.Partition; //导入依赖的package包/类
protected Partition[] getPartitions(GetPartitionsAction<T> getPartitionsAction) {

        if (extractor == null) {
            this.initExtractor(getPartitionsAction.getConfig());
        }

        return extractor.getPartitions(getPartitionsAction.getConfig());
    }
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public Seq<String> getPreferredLocations(Partition split) {
    initExtractorClient();

    List<String> locations = extractorClient.getPreferredLocations(split);
    if (locations == null || locations.isEmpty()) {
        return super.getPreferredLocations(split);
    }

    return asScalaBuffer(locations);

}
 

import org.apache.spark.Partition; //导入依赖的package包/类
@Override
public Iterator<T> compute(Partition split, TaskContext context) {

    initExtractorClient();

    extractorClient.initIterator(split, config.getValue());

    context.addTaskCompletionListener(new AbstractFunction1<TaskContext, BoxedUnit>() {

        @Override
        public BoxedUnit apply(TaskContext v1) {
            extractorClient.close();
            return null;
        }
    });

    java.util.Iterator<T> iterator = new java.util.Iterator<T>() {

        @Override
        public boolean hasNext() {
            return extractorClient.hasNext();
        }

        @Override
        public T next() {
            return extractorClient.next();
        }

        @Override
        public void remove() {
            throw new DeepIOException(
                    "Method not implemented (and won't be implemented anytime soon!!!)");
        }
    };

    return new InterruptibleIterator<>(context, asScalaIterator(iterator));

}