Running SQL to print to Std Out

[Mich Talebzadeh]

Hi,

This is the streaming program I have for trade prices following the doc for result set for tables

https://flink.apache.org/news/2017/03/29/table-sql-api-update.html

    val streamExecEnv = StreamExecutionEnvironment.getExecutionEnvironment
     val ds =  streamExecEnv
       .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))
    val splitStream = ds.map(new MapFunction[String, Tuple4[String, String, String, Float]] {
      override def map(value: String): Tuple4[String, String, String, Float] = {
        var cols = value.split(',')
        return (cols(0).toString, cols(1).toString, cols(2).toString, cols(3).toFloat)
      }
    })

    val tableEnv = TableEnvironment.getTableEnvironment(streamExecEnv)

    tableEnv.registerDataStream("priceTable", splitStream, 'key, 'ticker, 'timeissued, 'price)

    val result = tableEnv.scan("priceTable").filter('ticker.isNotNull).select('key, 'ticker, 'timeissued, 'price)
    val r = result.toDataStream[Row]
    r.print()

This compiles and runs but I do not see any ouput to screen.

This is the output from Flink GUI

I can verify that data being streamed in so there is no issue there. However, I don't see any output and Flink GUI does not look healthy (circles).

Appreciate any input.

Thanks,

[Hequn Cheng]

Hi Mich,

When you print to stdout on cluster, you have to look at the taskmanager .out file (also available in the UI).

Best, Hequn

On Tue, Aug 7, 2018 at 4:48 AM, Mich Talebzadeh <[hidden email]> wrote:

Hi,

This is the streaming program I have for trade prices following the doc for result set for tables

https://flink.apache.org/news/2017/03/29/table-sql-api-update.html

    val streamExecEnv = StreamExecutionEnvironment.getExecutionEnvironment
     val ds =  streamExecEnv
       .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))
    val splitStream = ds.map(new MapFunction[String, Tuple4[String, String, String, Float]] {
      override def map(value: String): Tuple4[String, String, String, Float] = {
        var cols = value.split(',')
        return (cols(0).toString, cols(1).toString, cols(2).toString, cols(3).toFloat)
      }
    })

    val tableEnv = TableEnvironment.getTableEnvironment(streamExecEnv)

    tableEnv.registerDataStream("priceTable", splitStream, 'key, 'ticker, 'timeissued, 'price)

    val result = tableEnv.scan("priceTable").filter('ticker.isNotNull).select('key, 'ticker, 'timeissued, 'price)
    val r = result.toDataStream[Row]
    r.print()

This compiles and runs but I do not see any ouput to screen.

This is the output from Flink GUI

I can verify that data being streamed in so there is no issue there. However, I don't see any output and Flink GUI does not look healthy (circles).

Appreciate any input.

Thanks,

[vino yang]

Hi Mich,

Hequn is correct, click on the “Task Managers” menu located on the left side of the Flink web UI, select a TM from the TM list, and then click on the “Stdout” option in the TM Details tab.

Thanks, vino.

2018-08-07 9:42 GMT+08:00 Hequn Cheng <[hidden email]>:

Hi Mich,

When you print to stdout on cluster, you have to look at the taskmanager .out file (also available in the UI).

Best, Hequn

On Tue, Aug 7, 2018 at 4:48 AM, Mich Talebzadeh <[hidden email]> wrote:

Hi,

This is the streaming program I have for trade prices following the doc for result set for tables

https://flink.apache.org/news/2017/03/29/table-sql-api-update.html

    val streamExecEnv = StreamExecutionEnvironment.getExecutionEnvironment
     val ds =  streamExecEnv
       .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))
    val splitStream = ds.map(new MapFunction[String, Tuple4[String, String, String, Float]] {
      override def map(value: String): Tuple4[String, String, String, Float] = {
        var cols = value.split(',')
        return (cols(0).toString, cols(1).toString, cols(2).toString, cols(3).toFloat)
      }
    })

    val tableEnv = TableEnvironment.getTableEnvironment(streamExecEnv)

    tableEnv.registerDataStream("priceTable", splitStream, 'key, 'ticker, 'timeissued, 'price)

    val result = tableEnv.scan("priceTable").filter('ticker.isNotNull).select('key, 'ticker, 'timeissued, 'price)
    val r = result.toDataStream[Row]
    r.print()

This compiles and runs but I do not see any ouput to screen.

This is the output from Flink GUI

I can verify that data being streamed in so there is no issue there. However, I don't see any output and Flink GUI does not look healthy (circles).

Appreciate any input.

Thanks,

[Mich Talebzadeh]

Awesome, thanks both.

I can see it now on task manager stdout

On Tue, 7 Aug 2018 at 03:15, vino yang <[hidden email]> wrote:

Hi Mich,

Hequn is correct, click on the “Task Managers” menu located on the left side of the Flink web UI, select a TM from the TM list, and then click on the “Stdout” option in the TM Details tab.

Thanks, vino.

2018-08-07 9:42 GMT+08:00 Hequn Cheng <[hidden email]>:

Hi Mich,

When you print to stdout on cluster, you have to look at the taskmanager .out file (also available in the UI).

Best, Hequn

On Tue, Aug 7, 2018 at 4:48 AM, Mich Talebzadeh <[hidden email]> wrote:

Hi,

This is the streaming program I have for trade prices following the doc for result set for tables

https://flink.apache.org/news/2017/03/29/table-sql-api-update.html

    val streamExecEnv = StreamExecutionEnvironment.getExecutionEnvironment
     val ds =  streamExecEnv
       .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))
    val splitStream = ds.map(new MapFunction[String, Tuple4[String, String, String, Float]] {
      override def map(value: String): Tuple4[String, String, String, Float] = {
        var cols = value.split(',')
        return (cols(0).toString, cols(1).toString, cols(2).toString, cols(3).toFloat)
      }
    })

    val tableEnv = TableEnvironment.getTableEnvironment(streamExecEnv)

    tableEnv.registerDataStream("priceTable", splitStream, 'key, 'ticker, 'timeissued, 'price)

    val result = tableEnv.scan("priceTable").filter('ticker.isNotNull).select('key, 'ticker, 'timeissued, 'price)
    val r = result.toDataStream[Row]
    r.print()

This compiles and runs but I do not see any ouput to screen.

This is the output from Flink GUI

I can verify that data being streamed in so there is no issue there. However, I don't see any output and Flink GUI does not look healthy (circles).

Appreciate any input.

Thanks,

[네이버]

On 6 Aug 2018, at 23:34, Mich Talebzadeh <[hidden email]> wrote:

Awesome, thanks both.

I can see it now on task manager stdout

<image.png>

On Tue, 7 Aug 2018 at 03:15, vino yang <[hidden email]> wrote:

Hi Mich,

Hequn is correct, click on the “Task Managers” menu located on the left side of the Flink web UI, select a TM from the TM list, and then click on the “Stdout” option in the TM Details tab.

Thanks, vino.

2018-08-07 9:42 GMT+08:00 Hequn Cheng <[hidden email]>:

Hi Mich,

When you print to stdout on cluster, you have to look at the taskmanager .out file (also available in the UI).

Best, Hequn

On Tue, Aug 7, 2018 at 4:48 AM, Mich Talebzadeh <[hidden email]> wrote:

Hi,

This is the streaming program I have for trade prices following the doc for result set for tables

https://flink.apache.org/news/2017/03/29/table-sql-api-update.html

    val streamExecEnv = StreamExecutionEnvironment.getExecutionEnvironment
     val ds =  streamExecEnv
       .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))
    val splitStream = ds.map(new MapFunction[String, Tuple4[String, String, String, Float]] {
      override def map(value: String): Tuple4[String, String, String, Float] = {
        var cols = value.split(',')
        return (cols(0).toString, cols(1).toString, cols(2).toString, cols(3).toFloat)
      }
    })

    val tableEnv = TableEnvironment.getTableEnvironment(streamExecEnv)

    tableEnv.registerDataStream("priceTable", splitStream, 'key, 'ticker, 'timeissued, 'price)

    val result = tableEnv.scan("priceTable").filter('ticker.isNotNull).select('key, 'ticker, 'timeissued, 'price)
    val r = result.toDataStream[Row]
    r.print()

This compiles and runs but I do not see any ouput to screen.

This is the output from Flink GUI

I can verify that data being streamed in so there is no issue there. However, I don't see any output and Flink GUI does not look healthy (circles).

Appreciate any input.

Thanks,