Java 类org.apache.spark.api.java.function.DoubleFunction 实例源码

public static void main(String[] args) throws Exception {
    if (args.length != 2) {
     throw new Exception("Usage BasicLoadJson [sparkMaster] [cassandraHost]");
    }
   String sparkMaster = args[0];
   String cassandraHost = args[1];
   SparkConf conf = new SparkConf(true)
     .set("spark.cassandra.connection.host", cassandraHost);

    JavaSparkContext sc = new JavaSparkContext(
     sparkMaster, "basicquerycassandra", conf);
   // entire table as an RDD
   // assumes your table test was created as CREATE TABLE test.kv(key text PRIMARY KEY, value int);
   JavaRDD<CassandraRow> data = javaFunctions(sc).cassandraTable("test" , "kv");
   // print some basic stats
   System.out.println(data.mapToDouble(new DoubleFunction<CassandraRow>() {
       public double call(CassandraRow row) {
         return row.getInt("value");
       }}).stats());
   // write some basic data to Cassandra
   ArrayList<KeyValue> input = new ArrayList<KeyValue>();
   input.add(KeyValue.newInstance("mostmagic", 3));
   JavaRDD<KeyValue> kvRDD = sc.parallelize(input);
   javaFunctions(kvRDD, KeyValue.class).saveToCassandra("test", "kv");
}

public static void main(String[] args) throws Exception {
    String master;
    if (args.length > 0) {
     master = args[0];
    } else {
        master = "local";
    }
    JavaSparkContext sc = new JavaSparkContext(
     master, "basicmaptodouble", System.getenv("SPARK_HOME"), System.getenv("JARS"));
   JavaRDD<Integer> rdd = sc.parallelize(Arrays.asList(1, 2, 3, 4));
   JavaDoubleRDD result = rdd.mapToDouble(
     new DoubleFunction<Integer>() {
       public double call(Integer x) {
         double y = (double) x;
         return y * y;
       }
     });
   System.out.println(StringUtils.join(result.collect(), ","));
}

/**
 * Action: Calculates the slope of a linear regression of every time series.
 *
 * Where: value = slope * timestamp
 * .. or:     y = slope * x
 *
 * @return the slopes (simple linear regression) of each an every time series in the RDD
 */
public JavaDoubleRDD getSlopes() {
    return this.mapToDouble((DoubleFunction<MetricTimeSeries>) mts -> {
                SimpleRegression regression = new SimpleRegression();
        mts.points().forEach(p -> regression.addData(p.getTimestamp(), p.getValue()));
                return regression.getSlope();
            }
    );
}

/**
 * Action: Counts the number of observations.
 *
 * @return the number of overall observations in all time series
 */
public long countObservations() {
    JavaDoubleRDD sizesRdd = this.mapToDouble(
            (DoubleFunction<MetricTimeSeries>) value -> (double) value.size());
    return sizesRdd.sum().longValue();
}