Struct spark_connect_rs::dataframe::StreamingQuery

pub struct StreamingQuery { /* private fields */ }

Represents the active streaming created from a start on the writer

This object is used to control and monitor the active stream

Implementations

impl StreamingQuery

pub fn new( spark_session: Box<SparkSession>, write_stream: WriteStreamOperationStartResult ) -> StreamingQuery

pub fn id(&self) -> String

pub fn run_id(&self) -> String

pub fn name(&self) -> Option<String>

pub async fn awaitTermination( self, timeout_ms: Option<i64> ) -> Result<bool, SparkError>

pub async fn lastProgress(self) -> Result<Value, SparkError>

Examples found in repository

examples/readstream.rs (line 34)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let spark: SparkSession =
        SparkSessionBuilder::remote("sc://127.0.0.1:15002/;user_id=stream_example")
            .build()
            .await?;

    let df = spark
        .readStream()
        .format("rate")
        .option("rowsPerSecond", "5")
        .load(None)?;

    let query = df
        .writeStream()
        .format("console")
        .queryName("example_stream")
        .outputMode(OutputMode::Append)
        .trigger(Trigger::ProcessingTimeInterval("1 seconds".to_string()))
        .start(None)
        .await?;

    // loop to get multiple progression stats
    for _ in 1..5 {
        thread::sleep(time::Duration::from_secs(5));
        let val = &query.clone().lastProgress().await?;
        println!("{}", val);
    }

    // stop the active stream
    query.stop().await?;

    Ok(())
}

pub async fn recentProgress(self) -> Result<Value, SparkError>

pub async fn isActive(self) -> Result<bool, SparkError>

pub async fn stop(self) -> Result<bool, SparkError>

Examples found in repository

examples/readstream.rs (line 39)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let spark: SparkSession =
        SparkSessionBuilder::remote("sc://127.0.0.1:15002/;user_id=stream_example")
            .build()
            .await?;

    let df = spark
        .readStream()
        .format("rate")
        .option("rowsPerSecond", "5")
        .load(None)?;

    let query = df
        .writeStream()
        .format("console")
        .queryName("example_stream")
        .outputMode(OutputMode::Append)
        .trigger(Trigger::ProcessingTimeInterval("1 seconds".to_string()))
        .start(None)
        .await?;

    // loop to get multiple progression stats
    for _ in 1..5 {
        thread::sleep(time::Duration::from_secs(5));
        let val = &query.clone().lastProgress().await?;
        println!("{}", val);
    }

    // stop the active stream
    query.stop().await?;

    Ok(())
}

pub async fn status(self) -> Result<StatusResult, SparkError>

Trait Implementations

impl Clone for StreamingQuery

fn clone(&self) -> StreamingQuery

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for StreamingQuery

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.