0

私は Spark ストリーミングの概念の初心者であり、ソケットからの Spark ストリーミングを理解しようとして、過去 2 日間から立ち往生しています。Spark はソケットに渡されたブロックを読み取ることができます。ただし、読み取りブロックに対して操作は実行されません。

ここにスパークコードがあります

package foo;
import java.io.File;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

import scala.Tuple2;

public class AppSocket {

public static void main(String[] args)
{

    SparkConf conf = new SparkConf().setAppName("KAFKA").setMaster("local");

    JavaStreamingContext jssc = new JavaStreamingContext(conf, new org.apache.spark.streaming.Duration(1000));

    JavaReceiverInputDStream<String> inputStream = jssc.socketTextStream("localhost", 33333);


    JavaPairDStream<String, Integer> mappedStream = inputStream.mapToPair(
        new PairFunction<String, String, Integer>() {

          public Tuple2<String, Integer> call(String i) {
             System.out.println(i);
            return new Tuple2<String, Integer>(i , 1);
          }
        });

    JavaPairDStream<String, Integer> reducedStream = mappedStream.reduceByKey(
      new Function2<Integer, Integer, Integer>() {

        public Integer call(Integer i1, Integer i2) {
          return i1 + i2;
        }
    });

    reducedStream.print();
    System.out.println("Testing........"+reducedStream.count());
    jssc.start();
    jssc.awaitTermination();


}

}

指定したポートに出力ストリームを作成するために netcat を実行しています

nc -lk 33333

出力ストリームを作成しようとしました。これが私のJavaコードです

    ServerSocket serverSocket = null;
    int portNumber = 33333;
    serverSocket = new ServerSocket(portNumber);

    System.out.println("Server Waiting.................");

    Socket clientSocket = serverSocket.accept();

    System.out.println("Server Connected!!!!!!!!!!!!!!!");


    // Wait for a message
    int countflag = 0;
    PrintWriter out = null;
    out = new PrintWriter(clientSocket.getOutputStream(), true);
    while(true)
    {
    Message message = consumer.receive(1000);

    if (message instanceof TextMessage) {
        TextMessage textMessage = (TextMessage) message;
        String text = textMessage.getText();
        System.out.println("Received: " + text);
        list.add(text);
        System.out.println(++countflag);
        if(list.size() > 50)
        {

        for(int i = 0; i < list.size() ; i++)
        {
       System.out.print(i);
        out.write(text);
        out.write("\n");
        out.flush();
        }

       list.clear();
        }

    } else {
        count++;

    }
    if(count > 100) break;
    }
    out.close();
    consumer.close();
    session.close();
    connection.close();

Spark はストリームで送信されたブロックを消費しますが、ストリーミングされたブロックで必要なアクションは実行しません。

Spark 出力コンソール

14/11/26 15:32:14 INFO MemoryStore: ensureFreeSpace(12) called with curMem=3521, maxMem=278302556
14/11/26 15:32:14 INFO MemoryStore: Block input-0-1417015934400 stored as bytes in memory (estimated size 12.0 B, free 265.4 MB)
14/11/26 15:32:14 INFO BlockManagerInfo: Added input-0-1417015934400 in memory on ip-10-0-1-56.ec2.internal:57275 (size: 12.0 B, free: 265.4 MB)
14/11/26 15:32:14 INFO BlockManagerMaster: Updated info of block input-0-1417015934400
14/11/26 15:32:14 WARN BlockManager: Block input-0-1417015934400 already exists on this machine; not re-adding it
14/11/26 15:32:14 INFO BlockGenerator: Pushed block input-0-1417015934400
14/11/26 15:32:15 INFO ReceiverTracker: Stream 0 received 1 blocks
14/11/26 15:32:15 INFO JobScheduler: Added jobs for time 1417015935000 ms

よろしくお願いします。前もって感謝します

4

1 に答える 1