import org.apache.spark.sql.streaming.StreamingQuery; //导入依赖的package包/类
private void start() {
	log.debug("-> start()");

	SparkSession spark = SparkSession.builder().appName("Read lines over a file stream").master("local")
			.getOrCreate();

	// @formatter:off
	Dataset<Row> df = spark
			.readStream()
			.format("text")
			.load(StreamingUtils.getInputDirectory());
	// @formatter:on

	StreamingQuery query = df.writeStream().outputMode(OutputMode.Update()).format("console").start();

	try {
		query.awaitTermination();
	} catch (StreamingQueryException e) {
		log.error("Exception while waiting for query to end {}.", e.getMessage(), e);
	}

	// In this case everything is a string
	df.show();
	df.printSchema();
}
 

import org.apache.spark.sql.streaming.StreamingQuery; //导入依赖的package包/类
private void start() {
    log.debug("-> start()");

    SparkSession spark = SparkSession.builder()
            .appName("Read lines over a file stream").master("local")
            .getOrCreate();

    Dataset<Row> df = spark
            .readStream()
            .format("text")
            .load(StreamingUtils.getInputDirectory());

    StreamingQuery query = df.writeStream().outputMode(OutputMode.Update())
            .format("console").start();

    try {
        query.awaitTermination();
    } catch (StreamingQueryException e) {
        log.error("Exception while waiting for query to end {}.", e.getMessage(), e);
    }

    // Never executed
    df.show();
    df.printSchema();
}
 

import org.apache.spark.sql.streaming.StreamingQuery; //导入依赖的package包/类
/**
 * Shutdown gracefully a streaming spark job and wait for specific amount of time before exiting.
 *
 * @param query
 * @param checkIntervalMillis whether the query has terminated or not within the checkIntervalMillis milliseconds.
 * @throws InterruptedException
 * @throws StreamingQueryException
 */
public static void shutdownGracefully(StreamingQuery query, long checkIntervalMillis) throws InterruptedException,
    StreamingQueryException {
  boolean isStopped = false;
  while (!isStopped) {
    isStopped = query.awaitTermination(checkIntervalMillis);
    if (!isStopped && sparkInfo.isShutdownRequested()) {
      LOG.info("Marker file has been removed, will attempt to stop gracefully the spark structured streaming query");
      query.stop();
    }
  }
}
 

import org.apache.spark.sql.streaming.StreamingQuery; //导入依赖的package包/类
public static void main(String args[]) throws StreamingQueryException {
    SparkSession spark = SparkSession
            .builder()
            .appName("JavaStructuredNetworkWordCount")
            .master("local")
            .config("spark.sql.shuffle.partitions", 8)
            .getOrCreate();

    // Create DataFrame representing the stream of input lines from connection to localhost:9999
    Dataset<Row> lines = spark
            .readStream()
            .format("socket")
            .option("host", "localhost")
            .option("port", 9999)
            .load();

   // Split the lines into words
    Dataset<String> words = lines
            .as(Encoders.STRING())
            .flatMap(
                    new FlatMapFunction<String, String>() {
                        @Override
                        public Iterator<String> call(String x) {
                            return Arrays.asList(x.split(" ")).iterator();
                        }
                    }, Encoders.STRING());

    // Generate running word count
    Dataset<Row> wordCounts = words.groupBy("value").count();

    // Start running the query that prints the running counts to the console
    StreamingQuery query = wordCounts.writeStream()
            .outputMode("complete")
            .format("console")
            .start();

    query.awaitTermination();
}
 

import org.apache.spark.sql.streaming.StreamingQuery; //导入依赖的package包/类
public StreamingQuery start(final DataStreamWriter<?> writer, final String path) {
    Function0<StreamingQuery> runFunction = new AbstractFunction0<StreamingQuery>() {
        @Override
        public StreamingQuery apply() {
            return writer.start(path);
        }
    };
    return harness.startTest(runFunction);
}
 

import org.apache.spark.sql.streaming.StreamingQuery; //导入依赖的package包/类
public void run(final DataStreamWriter<?> writer, final String path) {
    Function0<StreamingQuery> runFunction = new AbstractFunction0<StreamingQuery>() {
        @Override
        public StreamingQuery apply() {
            return writer.start(path);
        }
    };
    harness.runTest(runFunction);
}
 

import org.apache.spark.sql.streaming.StreamingQuery; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
//Read properties
Properties prop = PropertyFileReader.readPropertyFile();

//SparkSesion
SparkSession spark = SparkSession
	      .builder()
	      .appName("VideoStreamProcessor")
	      .master(prop.getProperty("spark.master.url"))
	      .getOrCreate();	

//directory to save image files with motion detected
final String processedImageDir = prop.getProperty("processed.output.dir");
logger.warn("Output directory for saving processed images is set to "+processedImageDir+". This is configured in processed.output.dir key of property file.");

//create schema for json message
StructType schema =  DataTypes.createStructType(new StructField[] { 
		DataTypes.createStructField("cameraId", DataTypes.StringType, true),
		DataTypes.createStructField("timestamp", DataTypes.TimestampType, true),
		DataTypes.createStructField("rows", DataTypes.IntegerType, true),
		DataTypes.createStructField("cols", DataTypes.IntegerType, true),
		DataTypes.createStructField("type", DataTypes.IntegerType, true),
		DataTypes.createStructField("data", DataTypes.StringType, true)
		});


//Create DataSet from stream messages from kafka
   Dataset<VideoEventData> ds = spark
     .readStream()
     .format("kafka")
     .option("kafka.bootstrap.servers", prop.getProperty("kafka.bootstrap.servers"))
     .option("subscribe", prop.getProperty("kafka.topic"))
     .option("kafka.max.partition.fetch.bytes", prop.getProperty("kafka.max.partition.fetch.bytes"))
     .option("kafka.max.poll.records", prop.getProperty("kafka.max.poll.records"))
     .load()
     .selectExpr("CAST(value AS STRING) as message")
     .select(functions.from_json(functions.col("message"),schema).as("json"))
     .select("json.*")
     .as(Encoders.bean(VideoEventData.class)); 
   
   //key-value pair of cameraId-VideoEventData
KeyValueGroupedDataset<String, VideoEventData> kvDataset = ds.groupByKey(new MapFunction<VideoEventData, String>() {
	@Override
	public String call(VideoEventData value) throws Exception {
		return value.getCameraId();
	}
}, Encoders.STRING());
	
//process
Dataset<VideoEventData> processedDataset = kvDataset.mapGroupsWithState(new MapGroupsWithStateFunction<String, VideoEventData, VideoEventData,VideoEventData>(){
	@Override
	public VideoEventData call(String key, Iterator<VideoEventData> values, GroupState<VideoEventData> state) throws Exception {
		logger.warn("CameraId="+key+" PartitionId="+TaskContext.getPartitionId());
		VideoEventData existing = null;
		//check previous state
		if (state.exists()) {
			existing = state.get();
		}
		//detect motion
		VideoEventData processed = VideoMotionDetector.detectMotion(key,values,processedImageDir,existing);
		
		//update last processed
		if(processed != null){
			state.update(processed);
		}
		return processed;
	}}, Encoders.bean(VideoEventData.class), Encoders.bean(VideoEventData.class));

//start
 StreamingQuery query = processedDataset.writeStream()
	      .outputMode("update")
	      .format("console")
	      .start();
 
 //await
    query.awaitTermination();
}
 

import org.apache.spark.sql.streaming.StreamingQuery; //导入依赖的package包/类
public static void main(String[] args) throws StreamingQueryException {
    //set log4j programmatically
    LogManager.getLogger("org.apache.spark").setLevel(Level.WARN);
    LogManager.getLogger("akka").setLevel(Level.ERROR);

    //configure Spark
    SparkConf conf = new SparkConf()
            .setAppName("kafka-structured")
            .setMaster("local[*]");

    //initialize spark session
    SparkSession sparkSession = SparkSession
            .builder()
            .config(conf)
            .getOrCreate();

    //reduce task number
    sparkSession.sqlContext().setConf("spark.sql.shuffle.partitions", "3");

    //data stream from kafka
    Dataset<Row> ds1 = sparkSession
            .readStream()
            .format("kafka")
            .option("kafka.bootstrap.servers", "localhost:9092")
            .option("subscribe", "mytopic")
            .option("startingOffsets", "earliest")
            .load();

    //start the streaming query
    sparkSession.udf().register("deserialize", (byte[] data) -> {
        GenericRecord record = recordInjection.invert(data).get();
        return RowFactory.create(record.get("str1").toString(), record.get("str2").toString(), record.get("int1"));

    }, DataTypes.createStructType(type.fields()));
    ds1.printSchema();
    Dataset<Row> ds2 = ds1
            .select("value").as(Encoders.BINARY())
            .selectExpr("deserialize(value) as rows")
            .select("rows.*");

    ds2.printSchema();

    StreamingQuery query1 = ds2
            .groupBy("str1")
            .count()
            .writeStream()
            .queryName("Test query")
            .outputMode("complete")
            .format("console")
            .start();

    query1.awaitTermination();

}
 

import org.apache.spark.sql.streaming.StreamingQuery; //导入依赖的package包/类
public static void main(String[] args) throws StreamingQueryException {
	System.setProperty("hadoop.home.dir", "C:\\softwares\\Winutils");
	SparkSession sparkSession = SparkSession.builder().master("local[*]").appName("structured Streaming Example")
			.config("spark.sql.warehouse.dir", "file:////C:/Users/sgulati/spark-warehouse").getOrCreate();

	Dataset<Row> inStream = sparkSession.readStream().format("socket").option("host", "10.204.136.223")
			.option("port", 9999).load();

	Dataset<FlightDetails> dsFlightDetails = inStream.as(Encoders.STRING()).map(x -> {
		ObjectMapper mapper = new ObjectMapper();
		return mapper.readValue(x, FlightDetails.class);

	}, Encoders.bean(FlightDetails.class));
	
	
	dsFlightDetails.createOrReplaceTempView("flight_details");
	
	Dataset<Row> avdFlightDetails = sparkSession.sql("select flightId, avg(temperature) from flight_details group by flightId");
	
	StreamingQuery query = avdFlightDetails.writeStream()
			  .outputMode("complete")
			  .format("console")
			  .start();

			query.awaitTermination();
	

}