|
|
Hi Lee,
I did try
Option 1:- it writes to CSV file only if I kill the running job. tEnv.toRetractStream(table, classOf[org.apache.flink.types.Row])
.writeAsCsv("/Users/kalit_000/Downloads/FlinkStreamAndSql/src/main/resources/csvOut3",
FileSystem.WriteMode.OVERWRITE,"~","|") OutPut:- 2> (true,180094108369013,John,Holland,c1ad7a1b73172ef67bd24820438f3f93,2019-07-15 22:48:40,travel,Satterfield-Lowe,81,39.015861,-119.883595)
Option 2:- I tried several options thought this workaround is kind of working but I need to replace brakcets,true etc....
import java.io.PrintStream
val fileOut = new PrintStream("/Users/kalit_000/Downloads/FlinkStreamAndSql/src/main/resources/csvOut2/out.txt")
System.setOut(fileOut)
tEnv.toRetractStream(table, classOf[org.apache.flink.types.Row]).print()
System.out.println(tEnv.toRetractStream(table, classOf[org.apache.flink.types.Row]).print()) Hi caizhi and kali:
I think this table should use toRetractStream instead of toAppendStream, and you should handle the retract messages. (If you just use distinct, the message should always be accumulate message)
Best, JingsongLee
------------------------------------------------------------------ Send Time:2019年7月16日(星期二) 09:52 Subject:Re: Stream to CSV Sink with SQL Distinct Values
Hi Kali,
Currently Flink treats all aggregate functions as retractable. As `distinct` is an aggregate function, it's considered by the planner that it might update or retract records (although from my perspective it won't...). Because csv table sink is an append only sink (it's hard to update what has been written in the middle of a file), the exception you mentioned occurs.
However, you can use `toAppendStream` method to change the retractable stream to an append only stream. For example, `tEnv.sqlQuery(query).distinct().toAppendStream[Row]` and then you can get an append only stream. You can then add csv sink to this stream.
Hi All,
I am trying to read data from kinesis stream and applying SQL transformation (distinct) and then tryting to write to CSV sink which is failinf due to this issue (org.apache.flink.table.api.TableException: AppendStreamTableSink requires that Table has only insert changes.) , full code is here ( https://github.com/kali786516/FlinkStreamAndSql/blob/614abfc100f74bd8bb7fadb926d946f16f6ef845/src/main/scala/com/aws/examples/kinesis/consumer/TransactionExample/KinesisConsumer.scala#L112).
can anyone help me moveforward on this issue?
Full Code:- // set up the streaming execution environment
val env = StreamExecutionEnvironment.createLocalEnvironment
//env.enableCheckpointing(10)
val tEnv = TableEnvironment.getTableEnvironment(env)
// Get AWS credentials
val credentialsProvider = new DefaultAWSCredentialsProviderChain
val credentials = credentialsProvider.getCredentials
// Configure Flink Kinesis consumer
val consumerConfig = new Properties
consumerConfig.put(AWSConfigConstants.AWS_REGION, "us-east-1")
consumerConfig.put(AWSConfigConstants.AWS_ACCESS_KEY_ID, credentials.getAWSAccessKeyId)
consumerConfig.put(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, credentials.getAWSSecretKey)
consumerConfig.put(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "TRIM_HORIZON")
// Create Kinesis stream
val kinesis = env.addSource(new FlinkKinesisConsumer("credittransactions2", new SimpleStringSchema(), consumerConfig))
val mapFunction: MapFunction[String, Tuple10[String, String, String,String,String,String,String,String,String,String]] =
new MapFunction[String, Tuple10[String, String, String,String,String,String,String,String,String,String]]() {
override def map(s: String): Tuple10[String, String, String,String,String,String,String,String,String,String] = {
val data = new Gson().fromJson(s, classOf[TransactionJsonClass])
val csvData = data.getCc_num+","+
data.getFirst+","+
data.getLast+","+
data.getTrans_num+","+
data.getTrans_time+","+
data.getCategory+","+
data.getMerchant+","+
data.getAmt+","+
data.getMerch_lat+","+
data.getMerch_long
//println(csvData)
val p:Array[String] = csvData.split(",")
var cc_num:String = p(0)
var first:String = p(1)
var last:String = p(2)
var trans_num:String = p(3)
var trans_time:String = p(4)
var category:String = p(5)
var merchant:String = p(6)
var amt:String = p(7)
var merch_lat:String = p(8)
var merch_long:String = p(9)
val creationDate: Time = new Time(System.currentTimeMillis())
return new Tuple10(cc_num, first, last,trans_num,trans_time,category,merchant,amt,merch_lat,merch_long)
}
}
val data = kinesis.map(mapFunction)
//data.print()
tEnv.registerDataStream("transactions",data,"cc_num,first_column,last_column,trans_num,trans_time,category_column,merchant_column,amt_column,merch_lat,merch_long")
val query = "SELECT distinct cc_num,first_column,last_column,trans_num,trans_time,category_column,merchant_column,amt_column,merch_lat,merch_long FROM transactions where cc_num not in ('cc_num')"
val table = tEnv.sqlQuery(query)
val table1 = table.distinct()
tEnv.registerTable("fromAnotherTable",table1)
table.printSchema()
val csvSink:TableSink[Row] = new CsvTableSink("/Users/kalit_000/Downloads/FlinkStreamAndSql/src/main/resources/csvOutput","~");
val fieldNames:Array[String] = Array("cc_num","first_column","last_column","trans_num","trans_time","category_column","merchant_column","amt_column","merch_lat","merch_long")
val fieldTypes:Array[TypeInformation[_]] = Array(
org.apache.flink.api.common.typeinfo.Types.STRING,
org.apache.flink.api.common.typeinfo.Types.STRING,
org.apache.flink.api.common.typeinfo.Types.STRING,
org.apache.flink.api.common.typeinfo.Types.STRING,
org.apache.flink.api.common.typeinfo.Types.STRING,
org.apache.flink.api.common.typeinfo.Types.STRING,
org.apache.flink.api.common.typeinfo.Types.STRING,
org.apache.flink.api.common.typeinfo.Types.STRING,
org.apache.flink.api.common.typeinfo.Types.STRING,
org.apache.flink.api.common.typeinfo.Types.STRING
)
tEnv.registerTableSink("s3csvTargetTransaction", fieldNames, fieldTypes, csvSink)
tEnv.sqlUpdate("INSERT INTO s3csvTargetTransaction SELECT cc_num,first_column,last_column,trans_num,trans_time,category_column,merchant_column,amt_column,merch_lat,merch_long from fromAnotherTable")
Thanks & Regards Sri Tummala
-- Thanks & Regards Sri Tummala
|
Locked
Screen Shot 2019-07-15 at 10.00.38 PM.png (105K)