import scala.reflect.ClassTag; //导入依赖的package包/类
/**
 * Convert a {@code MatrixObject} to a {@code RDD<String>} in IJV format.
 *
 * @param matrixObject
 *            the {@code MatrixObject}
 * @return the {@code MatrixObject} converted to a {@code RDD<String>}
 */
public static RDD<String> matrixObjectToRDDStringIJV(MatrixObject matrixObject) {

	// NOTE: The following works when called from Java but does not
	// currently work when called from Spark Shell (when you call
	// collect() on the RDD<String>).
	//
	// JavaRDD<String> javaRDD = jsc.parallelize(list);
	// RDD<String> rdd = JavaRDD.toRDD(javaRDD);
	//
	// Therefore, we call parallelize() on the SparkContext rather than
	// the JavaSparkContext to produce the RDD<String> for Scala.

	List<String> list = matrixObjectToListStringIJV(matrixObject);

	ClassTag<String> tag = scala.reflect.ClassTag$.MODULE$.apply(String.class);
	return sc().parallelize(JavaConversions.asScalaBuffer(list), sc().defaultParallelism(), tag);
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
/**
 * Convert a {@code FrameObject} to a {@code RDD<String>} in IJV format.
 *
 * @param frameObject
 *            the {@code FrameObject}
 * @return the {@code FrameObject} converted to a {@code RDD<String>}
 */
public static RDD<String> frameObjectToRDDStringIJV(FrameObject frameObject) {

	// NOTE: The following works when called from Java but does not
	// currently work when called from Spark Shell (when you call
	// collect() on the RDD<String>).
	//
	// JavaRDD<String> javaRDD = jsc.parallelize(list);
	// RDD<String> rdd = JavaRDD.toRDD(javaRDD);
	//
	// Therefore, we call parallelize() on the SparkContext rather than
	// the JavaSparkContext to produce the RDD<String> for Scala.

	List<String> list = frameObjectToListStringIJV(frameObject);

	ClassTag<String> tag = scala.reflect.ClassTag$.MODULE$.apply(String.class);
	return sc().parallelize(JavaConversions.asScalaBuffer(list), sc().defaultParallelism(), tag);
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
/**
 * Convert a {@code MatrixObject} to a {@code RDD<String>} in CSV format.
 *
 * @param matrixObject
 *            the {@code MatrixObject}
 * @return the {@code MatrixObject} converted to a {@code RDD<String>}
 */
public static RDD<String> matrixObjectToRDDStringCSV(MatrixObject matrixObject) {

	// NOTE: The following works when called from Java but does not
	// currently work when called from Spark Shell (when you call
	// collect() on the RDD<String>).
	//
	// JavaRDD<String> javaRDD = jsc.parallelize(list);
	// RDD<String> rdd = JavaRDD.toRDD(javaRDD);
	//
	// Therefore, we call parallelize() on the SparkContext rather than
	// the JavaSparkContext to produce the RDD<String> for Scala.

	List<String> list = matrixObjectToListStringCSV(matrixObject);

	ClassTag<String> tag = scala.reflect.ClassTag$.MODULE$.apply(String.class);
	return sc().parallelize(JavaConversions.asScalaBuffer(list), sc().defaultParallelism(), tag);
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
/**
 * Convert a {@code FrameObject} to a {@code RDD<String>} in CSV format.
 *
 * @param frameObject
 *            the {@code FrameObject}
 * @param delimiter
 *            the delimiter
 * @return the {@code FrameObject} converted to a {@code RDD<String>}
 */
public static RDD<String> frameObjectToRDDStringCSV(FrameObject frameObject, String delimiter) {

	// NOTE: The following works when called from Java but does not
	// currently work when called from Spark Shell (when you call
	// collect() on the RDD<String>).
	//
	// JavaRDD<String> javaRDD = jsc.parallelize(list);
	// RDD<String> rdd = JavaRDD.toRDD(javaRDD);
	//
	// Therefore, we call parallelize() on the SparkContext rather than
	// the JavaSparkContext to produce the RDD<String> for Scala.

	List<String> list = frameObjectToListStringCSV(frameObject, delimiter);

	ClassTag<String> tag = scala.reflect.ClassTag$.MODULE$.apply(String.class);
	return sc().parallelize(JavaConversions.asScalaBuffer(list), sc().defaultParallelism(), tag);
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public JavaStreamingContext create() {
  sparkConf.set("spark.streaming.kafka.maxRatePerPartition", String.valueOf(maxRatePerPartition));
  JavaStreamingContext result = new JavaStreamingContext(sparkConf, new Duration(duration));
  Map<String, String> props = new HashMap<>();
  if (!autoOffsetValue.isEmpty()) {
    props.put(AbstractStreamingBinding.AUTO_OFFSET_RESET, autoOffsetValue);
  }
  logMessage("topic list " + topic, isRunningInMesos);
  logMessage("Auto offset reset is set to " + autoOffsetValue, isRunningInMesos);
  props.putAll(extraKafkaConfigs);
  for (Map.Entry<String, String> map : props.entrySet()) {
    logMessage(Utils.format("Adding extra kafka config, {}:{}", map.getKey(), map.getValue()), isRunningInMesos);
  }
  props.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
  props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
  props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
  JavaPairInputDStream<byte[], byte[]> dStream;
  if (offsetHelper.isSDCCheckPointing()) {
    JavaInputDStream stream =
        KafkaUtils.createDirectStream(
            result,
            byte[].class,
            byte[].class,
            Tuple2.class,
            props,
            MaprStreamsOffsetManagerImpl.get().getOffsetForDStream(topic, numberOfPartitions),
            MESSAGE_HANDLER_FUNCTION
        );
    ClassTag<byte[]> byteClassTag = scala.reflect.ClassTag$.MODULE$.apply(byte[].class);
    dStream = JavaPairInputDStream.fromInputDStream(stream.inputDStream(), byteClassTag, byteClassTag);
  } else {
    dStream =
        KafkaUtils.createDirectStream(result, byte[].class, byte[].class,
            props, new HashSet<>(Arrays.asList(topic.split(","))));
  }
  Driver$.MODULE$.foreach(dStream.dstream(), MaprStreamsOffsetManagerImpl.get());
  return result;
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
@Override
public RDD<Tuple> convert(List<RDD<Tuple>> predecessors,
        PODistinct poDistinct) throws IOException {
    SparkUtil.assertPredecessorSize(predecessors, poDistinct, 1);
    RDD<Tuple> rdd = predecessors.get(0);

    ClassTag<Tuple2<Tuple, Object>> tuple2ClassManifest = SparkUtil
            .<Tuple, Object> getTuple2Manifest();

    RDD<Tuple2<Tuple, Object>> rddPairs = rdd.map(TO_KEY_VALUE_FUNCTION,
            tuple2ClassManifest);
    PairRDDFunctions<Tuple, Object> pairRDDFunctions
      = new PairRDDFunctions<Tuple, Object>(
            rddPairs, SparkUtil.getManifest(Tuple.class),
            SparkUtil.getManifest(Object.class), null);
    int parallelism = SparkUtil.getParallelism(predecessors, poDistinct);
    return pairRDDFunctions.reduceByKey(MERGE_VALUES_FUNCTION, parallelism)
            .map(TO_VALUE_FUNCTION, SparkUtil.getManifest(Tuple.class));
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
public MizoRDD(SparkContext context, IMizoRDDConfig config, ClassTag<TReturn> classTag) {
    super(context, new ArrayBuffer<>(), classTag);

    if (!Strings.isNullOrEmpty(config.logConfigPath())) {
        PropertyConfigurator.configure(config.logConfigPath());
    }

    this.config = config;
    this.regionsPaths = getRegionsPaths(config.regionDirectoriesPath());
    this.relationTypes = loadRelationTypes(config.titanConfigPath());
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
@Override
public <T> T deserialize(final ByteBuffer byteBuffer, final ClassLoader classLoader, final ClassTag<T> classTag) {
    this.input.setBuffer(byteBuffer.array());
    return this.gryoSerializer.getGryoPool().readWithKryo(kryo -> {
        kryo.setClassLoader(classLoader);
        return (T) kryo.readClassAndObject(this.input);
    });
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
public static <T> DStream<Tuple2<Integer, Iterable<Long>>>  getPartitionOffset(
    DStream<MessageAndMetadata<T>> unionStreams, Properties props) {
  ClassTag<MessageAndMetadata<T>> messageMetaClassTag = 
      ScalaUtil.<T>getMessageAndMetadataClassTag();
  JavaDStream<MessageAndMetadata<T>> javaDStream = 
      new JavaDStream<MessageAndMetadata<T>>(unionStreams, messageMetaClassTag);
  JavaPairDStream<Integer, Iterable<Long>> partitonOffset = getPartitionOffset(javaDStream, props);
  return partitonOffset.dstream();
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
@SuppressWarnings("deprecation")
public static void persists(DStream<Tuple2<Integer, Iterable<Long>>> partitonOffset, Properties props) {
  ClassTag<Tuple2<Integer, Iterable<Long>>> tuple2ClassTag = 
      ScalaUtil.<Integer, Iterable<Long>>getTuple2ClassTag();
  JavaDStream<Tuple2<Integer, Iterable<Long>>> jpartitonOffset = 
      new JavaDStream<Tuple2<Integer, Iterable<Long>>>(partitonOffset, tuple2ClassTag);
  jpartitonOffset.foreachRDD(new VoidFunction<JavaRDD<Tuple2<Integer, Iterable<Long>>>>() {
    @Override
    public void call(JavaRDD<Tuple2<Integer, Iterable<Long>>> po) throws Exception {
      List<Tuple2<Integer, Iterable<Long>>> poList = po.collect();
      doPersists(poList, props);
    }
  });
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
public MatrixMultiplicationRDD(final RDD<?> oneParent, final ClassTag<MatrixStore<N>> evidence$2) {
    super(oneParent, evidence$2);
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
public NumberRDD(final RDD<?> oneParent, final ClassTag<N> evidence$2) {
    super(oneParent, evidence$2);
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
public OtherBlockMatrixRDD(final RDD<?> oneParent, final ClassTag<MatrixStore<N>> evidence$2) {
    super(oneParent, evidence$2);
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
@Override
public <T> ByteBuffer serialize(final T t, final ClassTag<T> classTag) {
    this.gryoSerializer.getGryoPool().writeWithKryo(kryo -> kryo.writeClassAndObject(this.output, t));
    return ByteBuffer.wrap(this.output.getBuffer());
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
@Override
public <T> SerializationStream writeObject(final T t, final ClassTag<T> classTag) {
    this.gryoSerializer.getGryoPool().writeWithKryo(kryo -> kryo.writeClassAndObject(this.output, t));
    return this;
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
public Graph<RyaTypeWritable, RyaTypeWritable> createGraph(SparkContext sc, Configuration conf) throws IOException, AccumuloSecurityException{
    StorageLevel storageLvl1 = StorageLevel.MEMORY_ONLY();
    StorageLevel storageLvl2 = StorageLevel.MEMORY_ONLY();
    ClassTag<RyaTypeWritable> RTWTag = ClassTag$.MODULE$.apply(RyaTypeWritable.class);
    RyaTypeWritable rtw = null;
    RDD<Tuple2<Object, RyaTypeWritable>> vertexRDD = getVertexRDD(sc, conf);

    RDD<Tuple2<Object, Edge>> edgeRDD = getEdgeRDD(sc, conf);
    JavaRDD<Tuple2<Object, Edge>> jrddTuple = edgeRDD.toJavaRDD();
    JavaRDD<Edge<RyaTypeWritable>> jrdd = jrddTuple.map(tuple -> tuple._2);

    RDD<Edge<RyaTypeWritable>> goodERDD = JavaRDD.toRDD(jrdd);

    return Graph.apply(vertexRDD, goodERDD, rtw, storageLvl1, storageLvl2, RTWTag, RTWTag);
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public JavaStreamingContext create() {
  sparkConf.set("spark.streaming.kafka.maxRatePerPartition", String.valueOf(maxRatePerPartition));
  JavaStreamingContext result = new JavaStreamingContext(sparkConf, new Duration(duration));
  Map<String, String> props = new HashMap<>();
  props.putAll(extraKafkaConfigs);
  for (Map.Entry<String, String> map : props.entrySet()) {
    logMessage(Utils.format("Adding extra kafka config, {}:{}", map.getKey(), map.getValue()), isRunningInMesos);
  }
  props.put("metadata.broker.list", metaDataBrokerList);
  props.put(GROUP_ID_KEY, groupId);
  if (!autoOffsetValue.isEmpty()) {
    autoOffsetValue = getConfigurableAutoOffsetResetIfNonEmpty(autoOffsetValue);
    props.put(AUTO_OFFSET_RESET, autoOffsetValue);
  }
  logMessage("Meta data broker list " + metaDataBrokerList, isRunningInMesos);
  logMessage("Topic is " + topic, isRunningInMesos);
  logMessage("Auto offset reset is set to " + autoOffsetValue, isRunningInMesos);
  JavaPairInputDStream<byte[], byte[]> dStream;
  if (offsetHelper.isSDCCheckPointing()) {
    JavaInputDStream<Tuple2<byte[], byte[]>> stream =
        KafkaUtils.createDirectStream(
            result,
            byte[].class,
            byte[].class,
            DefaultDecoder.class,
            DefaultDecoder.class,
            (Class<Tuple2<byte[], byte[]>>) ((Class)(Tuple2.class)),
            props,
            KafkaOffsetManagerImpl.get().getOffsetForDStream(topic, numberOfPartitions),
            MESSAGE_HANDLER_FUNCTION
        );
    ClassTag<byte[]> byteClassTag = scala.reflect.ClassTag$.MODULE$.apply(byte[].class);
    dStream = JavaPairInputDStream.fromInputDStream(stream.inputDStream(), byteClassTag, byteClassTag);
  } else {
    dStream =
        KafkaUtils.createDirectStream(result, byte[].class, byte[].class, DefaultDecoder.class, DefaultDecoder.class,
            props, new HashSet<>(Arrays.asList(topic.split(","))));
  }
  Driver$.MODULE$.foreach(dStream.dstream(), KafkaOffsetManagerImpl.get());
  return result;
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
@Override
public ClassTag<T> classTag() {
    return ClassTag$.MODULE$.<T>apply(((BaseConfig<T,BaseConfig>)((DeepRDD) this.rdd()).config.value())
            .getEntityClass());
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
public static <T> ClassTag<T> getManifest(Class<T> clazz) {
    return ClassTag$.MODULE$.apply(clazz);
}
 

import scala.reflect.ClassTag; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public static <K,V> ClassTag<Tuple2<K, V>> getTuple2Manifest() {
    return (ClassTag<Tuple2<K, V>>)(Object)getManifest(Tuple2.class);
}