Spark - can a MultiMap be converted to a DataFrame in JAVA

Question

I am trying to convert a MultiMap from billions of data values ​​to a Spark DataFrame to run calculations and then write the results to a cassandra table.

I generate multimap from the following query and cassandra loop. I would be glad to accept the suggestions if there was a better way to get and process this data in a DataFrame, as I am with the loop.

Updated code with the answer:

//Build ResultSet from cassandra query for data manipulation.
        Statement stmt = new SimpleStatement("SELECT \"Power\",\"Bandwidth\",\"Start_Frequency\" FROM \"SB1000_49552019\".\"Measured_Value\";");
        //Statement stmt = new SimpleStatement("SELECT power, bandwidth, start_frequency FROM model.reports;");
        stmt.setFetchSize(1000);
        ResultSet results = session.execute(stmt);

// Get the Variables from each Row of Cassandra Data        
 Multimap<Double, Float> data = LinkedListMultimap.create();
        for (Row row : results){       
           // Column Names in Cassandra (Case Sensitive)
           start_frequency = row.getDouble("Start_Frequency");
           power = row.getFloat("Power");
           bandwidth = row.getDouble("Bandwidth"); 

// Create Channel Power Buckets, place information into prepared statement binding, write to cassandra.            
                for(channel = 1.6000E8; channel <= channel_end;  ){ 
                    if( (channel >= start_frequency) && (channel <= (start_frequency + bandwidth)) ) {     
                     data.put(channel, power);
                    }  // end if
                    channel+=increment;
                }  // end for      
        } // end "row" for

// Create Spark List for DataFrame        
        List<Value> values = data.asMap().entrySet()
            .stream()
            .flatMap(x -> x.getValue()
                    .stream()
                    .map(y -> new Value(x.getKey(), y)))
            .collect(Collectors.toList());

// Create DataFrame and Calculate Results
    sqlContext.createDataFrame(sc.parallelize(values), Value.class).groupBy(col("channel"))
        .agg(min("power"), max("power"), avg("power"))
        .write().mode(SaveMode.Append)      
        .option("table", "results")
        .option("keyspace", "model")
        .format("org.apache.spark.sql.cassandra").save();

    } // end session
} // End Compute 

public class Value implements Serializable {
    public Value(Double channel, Float power) {
        this.channel = channel;
        this.power = power;
    }
    Double channel;
    Float power;

    public void setChannel(Double channel) {
        this.channel = channel;
    }
    public void setPower(Float power) {
        this.power = power;
    }
    public Double getChannel() {
        return channel;
    }
    public Float getPower() {
        return power;
    }

    @Override
    public String toString() {
        return "[" +channel +","+power+"]";
    }
}

There are {Double} = [Float] types in the multimap sample, where there can be several Float elements for each Double

Example

{1.50E8=[10, 20], 1.51E8=[-10, -13, -14, -15], 1.52E8=[-10, -11]

I need to use a spark to get min, max, the average value for each of them. For example, for the first 1.50ED there will be min 10, max 20, avg 15.

, , :

queryMV.groupBy(col("channel"))
.agg(min("power"), max("power"), avg("power"))
.write().mode(SaveMode.Append)      
.option("table", "results")
.option("keyspace", "model")
.format("org.apache.spark.sql.cassandra").save();

, DataFrame JAVA. , .

, , for , , , / , , . multimap - .

Answer 1

, Java parallelize T, <<22 > Tuple<K, V>. . createDataFrame RDD Scala Seq ( bean, StructType).

com.google.common.collect.ImmutableEntry , Java, @Pankaj Arora, Java, , Java.

public class Value implements Serializable {
    public Value(Double a, Float b) {
        this.a = a;
        this.b = b;
    }
    Double a;
    Float b;

    public void setA(Double a) {
        this.a = a;
    }
    public void setB(Float b) {
        this.b = b;
    }
    public Double getA() {
        return a;
    }
    public Float getB() {
        return b;
    }

    public String toString() {
        return "[" +a +","+b+"]";
    }
}


    Multimap<Double, Float> data = LinkedListMultimap.create();
    data.put(1d, 1f);
    data.put(1d, 2f);
    data.put(2d, 3f);

    List<Value> values = data.asMap().entrySet()
            .stream()
            .flatMap(x -> x.getValue()
                    .stream()
                    .map(y -> new Value(x.getKey(), y)))
            .collect(Collectors.toList());

    sqlContext.createDataFrame(sc.parallelize(values), Value.class).show();

, ( ) .

Answer 2

case class Output(a : Double ,b : Int )
val input = Map(1.50E8-> List(10, 20) ,  1.51E8-> List( -10, -13, -14, -15 ), 1.52E8-> List(-10, -11)).toArray
val inputRdd = sc.parallelize(input)
val queryMV = inputRdd.flatMap(x=> x._2.map(y=> Output(x._1, y))).toDF