Struct spark_connect_rs::dataframe::DataStreamWriter

pub struct DataStreamWriter { /* private fields */ }

DataStreamWriter provides the ability to output a StreamingQuery which can then be used to monitor the active stream

Implementations

impl DataStreamWriter

pub fn new(dataframe: DataFrame) -> DataStreamWriter

Create a new DataStreamWriter from a provided streaming DataFrame

Defaults

• format: None,
• output_mode: OutputMode,
• query_name: None,
• trigger: Trigger,
• partition_by: vec![],
• write_options: HashMap::new()

pub fn format(self, format: &str) -> DataStreamWriter

Target format to output the StreamingQuery

Examples found in repository

examples/readstream.rs (line 24)

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 fn outputMode(self, outputMode: OutputMode) -> DataStreamWriter

Specifies the behavior when data or table already exists

Arguments:

• output_mode: OutputMode enum

Examples found in repository

examples/readstream.rs (line 26)

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 fn partitionBy<'a, I>(self, cols: I) -> DataStreamWriter

where I: IntoIterator<Item = &'a str>,

Partitions the output by the given columns on the file system

pub fn option(self, key: &str, value: &str) -> DataStreamWriter

Add an input option for the underlying data source

pub fn options<I, K, V>(self, options: I) -> DataStreamWriter

where I: IntoIterator<Item = (K, V)>, K: AsRef<str>, V: AsRef<str>,

Set many input options based on an iterator of (key/value pairs) for the underlying data source

pub fn queryName(self, name: &str) -> DataStreamWriter

Examples found in repository

examples/readstream.rs (line 25)

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 fn trigger(self, trigger: Trigger) -> DataStreamWriter

Query trigger for data to be processed by

Examples found in repository

examples/readstream.rs (line 27)

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 start( self, path: Option<&str> ) -> Result<StreamingQuery, SparkError>

Start a streaming job to save the contents of the StreamingQuery to a data source.

The data source is specified by the format and a set of options.

Examples found in repository

examples/readstream.rs (line 28)

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 toTable( self, tableName: &str ) -> Result<StreamingQuery, SparkError>

Start a streaming job to save the contents of the StreamingQuery to a table.

Trait Implementations

impl Clone for DataStreamWriter

fn clone(&self) -> DataStreamWriter

Returns a copy of the value.

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

Performs copy-assignment from source.

impl Debug for DataStreamWriter

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

Formats the value using the given formatter.