flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Aljoscha Krettek <aljos...@apache.org>
Subject Re: Help with Flink experimental Table API
Date Sun, 14 Jun 2015 17:41:32 GMT
Hi,
sorry, my mail client sent before I was done.

I think the problem is that the Scala compiler derives a wrong type for
this statement:
val table = env.fromElements((123, "a"), (234, "b"), (345, "c"), (null,
"d")).toTable

Because of the null value it derives (Any, String) as the type if you do it
like this, I think it should work:
val table = env.fromElements[(Integer, String)]((123, "a"), (234, "b"),
(345, "c"), (null, "d")).toTable

I used Integer instead of Int because Scala will complain that null is not
a valid value for Int otherwise.

Cheers,
Aljoscha


On Sun, 14 Jun 2015 at 19:34 Aljoscha Krettek <aljoscha@apache.org> wrote:

> Hi,
> I think the problem is that the Scala compiler derives a wrong type for
> this statement:
>
>
>
> On Sun, 14 Jun 2015 at 18:28 Shiti Saxena <ssaxena.ece@gmail.com> wrote:
>
>> Hi Aljoscha,
>>
>> I created the issue FLINK-2210
>> <https://issues.apache.org/jira/browse/FLINK-2210> for aggregate on
>> null. I made changes to ExpressionAggregateFunction to handle ignore null
>> values. But I am unable to create a Table with null values in tests.
>>
>> The code I used is,
>>
>> def testAggregationWithNull(): Unit = {
>>
>>     val env = ExecutionEnvironment.getExecutionEnvironment
>>     val table = env.fromElements((123, "a"), (234, "b"), (345, "c"),
>> (null, "d")).toTable
>>
>>     val total = table.select('_1.sum).collect().head.productElement(0)
>>     assertEquals(total, 702)
>>   }
>>
>> and the error i get is,
>>
>> org.apache.flink.api.table.ExpressionException: Invalid expression
>> "('_1).sum": Unsupported type GenericType<java.lang.Object> for aggregation
>> ('_1).sum. Only numeric data types supported.
>> at
>> org.apache.flink.api.table.expressions.analysis.TypeCheck.apply(TypeCheck.scala:50)
>> at
>> org.apache.flink.api.table.expressions.analysis.TypeCheck.apply(TypeCheck.scala:31)
>> at
>> org.apache.flink.api.table.trees.Analyzer$$anonfun$analyze$1.apply(Analyzer.scala:34)
>> at
>> org.apache.flink.api.table.trees.Analyzer$$anonfun$analyze$1.apply(Analyzer.scala:31)
>> at scala.collection.immutable.List.foreach(List.scala:318)
>> at org.apache.flink.api.table.trees.Analyzer.analyze(Analyzer.scala:31)
>> at org.apache.flink.api.table.Table$$anonfun$1.apply(Table.scala:59)
>> at org.apache.flink.api.table.Table$$anonfun$1.apply(Table.scala:59)
>> at
>> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
>> at
>> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
>> at
>> scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
>> at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:34)
>> at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
>> at scala.collection.AbstractTraversable.map(Traversable.scala:105)
>> at org.apache.flink.api.table.Table.select(Table.scala:59)
>> at
>> org.apache.flink.api.scala.table.test.AggregationsITCase.testAggregationWithNull(AggregationsITCase.scala:135)
>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>> at
>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
>> at
>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>> at
>> org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:47)
>> at
>> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
>> at
>> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:44)
>> at
>> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
>> at
>> org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
>> at
>> org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
>> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
>> at org.junit.rules.RunRules.evaluate(RunRules.java:20)
>> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:271)
>> at
>> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:70)
>> at
>> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:50)
>> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:238)
>> at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:63)
>> at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:236)
>> at org.junit.runners.ParentRunner.access$000(ParentRunner.java:53)
>> at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:229)
>> at org.junit.runners.ParentRunner.run(ParentRunner.java:309)
>> at org.junit.runners.Suite.runChild(Suite.java:127)
>> at org.junit.runners.Suite.runChild(Suite.java:26)
>> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:238)
>> at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:63)
>> at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:236)
>> at org.junit.runners.ParentRunner.access$000(ParentRunner.java:53)
>> at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:229)
>> at
>> org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
>> at
>> org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
>> at org.junit.runners.ParentRunner.run(ParentRunner.java:309)
>> at org.junit.runner.JUnitCore.run(JUnitCore.java:160)
>> at
>> com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:78)
>> at
>> com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:212)
>> at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:68)
>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>> at
>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
>> at com.intellij.rt.execution.application.AppMain.main(AppMain.java:140)
>>
>>
>> The ExecutionEnvironment.fromCollection method also throws an error when
>> the collection contains a null.
>>
>> Could you please point out what I am doing wrong? How do we create a
>> Table with null values?
>>
>> In our application, we load a file and transform each line into a Row
>> resulting in a DataSet[Row]. This DataSet[Row] is then converted into
>> Table. Should I use the same approach for the test case?
>>
>>
>> Thanks,
>> Shiti
>>
>>
>>
>>
>>
>>
>>
>>
>>
>> On Sun, Jun 14, 2015 at 4:10 PM, Shiti Saxena <ssaxena.ece@gmail.com>
>> wrote:
>>
>>> I'll do the fix
>>>
>>> On Sun, Jun 14, 2015 at 12:42 AM, Aljoscha Krettek <aljoscha@apache.org>
>>> wrote:
>>>
>>>> I merged your PR for the RowSerializer. Teaching the aggregators to
>>>> deal with null values should be a very simple fix in
>>>> ExpressionAggregateFunction.scala. There it is simply always aggregating
>>>> the values without checking whether they are null. If you want you can also
>>>> fix that or I can quickly fix it.
>>>>
>>>> On Thu, 11 Jun 2015 at 10:40 Aljoscha Krettek <aljoscha@apache.org>
>>>> wrote:
>>>>
>>>>> Cool, good to hear.
>>>>>
>>>>> The PojoSerializer already handles null fields. The RowSerializer can
>>>>> be modified in pretty much the same way. So you should start by looking
at
>>>>> the copy()/serialize()/deserialize() methods of PojoSerializer and then
>>>>> modify RowSerializer in a similar way.
>>>>>
>>>>> You can also send me a private mail if you want more in-depth
>>>>> explanations.
>>>>>
>>>>> On Thu, 11 Jun 2015 at 09:33 Till Rohrmann <trohrmann@apache.org>
>>>>> wrote:
>>>>>
>>>>>> Hi Shiti,
>>>>>>
>>>>>> here is the issue [1].
>>>>>>
>>>>>> Cheers,
>>>>>> Till
>>>>>>
>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-2203
>>>>>>
>>>>>> On Thu, Jun 11, 2015 at 8:42 AM Shiti Saxena <ssaxena.ece@gmail.com>
>>>>>> wrote:
>>>>>>
>>>>>>> Hi Aljoscha,
>>>>>>>
>>>>>>> Could you please point me to the JIRA tickets? If you could provide
>>>>>>> some guidance on how to resolve these, I will work on them and
raise a
>>>>>>> pull-request.
>>>>>>>
>>>>>>> Thanks,
>>>>>>> Shiti
>>>>>>>
>>>>>>> On Thu, Jun 11, 2015 at 11:31 AM, Aljoscha Krettek <
>>>>>>> aljoscha@apache.org> wrote:
>>>>>>>
>>>>>>>> Hi,
>>>>>>>> yes, I think the problem is that the RowSerializer does not
support
>>>>>>>> null-values. I think we can add support for this, I will
open a Jira issue.
>>>>>>>>
>>>>>>>> Another problem I then see is that the aggregations can not
>>>>>>>> properly deal with null-values. This would need separate
support.
>>>>>>>>
>>>>>>>> Regards,
>>>>>>>> Aljoscha
>>>>>>>>
>>>>>>>> On Thu, 11 Jun 2015 at 06:41 Shiti Saxena <ssaxena.ece@gmail.com>
>>>>>>>> wrote:
>>>>>>>>
>>>>>>>>> Hi,
>>>>>>>>>
>>>>>>>>> In our project, we are using the Flink Table API and
are facing
>>>>>>>>> the following issues,
>>>>>>>>>
>>>>>>>>> We load data from a CSV file and create a DataSet[Row].
The CSV
>>>>>>>>> file can also have invalid entries in some of the fields
which we replace
>>>>>>>>> with null when building the DataSet[Row].
>>>>>>>>>
>>>>>>>>> This DataSet[Row] is later on transformed to Table whenever
>>>>>>>>> required and specific operation such as select or aggregate,
etc are
>>>>>>>>> performed.
>>>>>>>>>
>>>>>>>>> When a null value is encountered, we get a null pointer
exception
>>>>>>>>> and the whole job fails. (We can see this by calling
collect on the
>>>>>>>>> resulting DataSet).
>>>>>>>>>
>>>>>>>>> The error message is similar to,
>>>>>>>>>
>>>>>>>>> Job execution failed.
>>>>>>>>> org.apache.flink.runtime.client.JobExecutionException:
Job
>>>>>>>>> execution failed.
>>>>>>>>> at
>>>>>>>>> org.apache.flink.runtime.jobmanager.JobManager$$anonfun$receiveWithLogMessages$1.applyOrElse(JobManager.scala:315)
>>>>>>>>> at
>>>>>>>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
>>>>>>>>> at
>>>>>>>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
>>>>>>>>> at
>>>>>>>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
>>>>>>>>> at
>>>>>>>>> org.apache.flink.runtime.ActorLogMessages$$anon$1.apply(ActorLogMessages.scala:43)
>>>>>>>>> at
>>>>>>>>> org.apache.flink.runtime.ActorLogMessages$$anon$1.apply(ActorLogMessages.scala:29)
>>>>>>>>> at
>>>>>>>>> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
>>>>>>>>> at
>>>>>>>>> org.apache.flink.runtime.ActorLogMessages$$anon$1.applyOrElse(ActorLogMessages.scala:29)
>>>>>>>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
>>>>>>>>> at
>>>>>>>>> org.apache.flink.runtime.jobmanager.JobManager.aroundReceive(JobManager.scala:94)
>>>>>>>>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
>>>>>>>>> at akka.actor.ActorCell.invoke(ActorCell.scala:487)
>>>>>>>>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
>>>>>>>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
>>>>>>>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
>>>>>>>>> at
>>>>>>>>> scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
>>>>>>>>> at
>>>>>>>>> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
>>>>>>>>> at
>>>>>>>>> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
>>>>>>>>> at
>>>>>>>>> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
>>>>>>>>> Caused by: java.lang.NullPointerException
>>>>>>>>> at
>>>>>>>>> org.apache.flink.api.common.typeutils.base.IntSerializer.serialize(IntSerializer.java:63)
>>>>>>>>> at
>>>>>>>>> org.apache.flink.api.common.typeutils.base.IntSerializer.serialize(IntSerializer.java:27)
>>>>>>>>> at
>>>>>>>>> org.apache.flink.api.table.typeinfo.RowSerializer.serialize(RowSerializer.scala:80)
>>>>>>>>> at
>>>>>>>>> org.apache.flink.api.table.typeinfo.RowSerializer.serialize(RowSerializer.scala:28)
>>>>>>>>> at
>>>>>>>>> org.apache.flink.runtime.plugable.SerializationDelegate.write(SerializationDelegate.java:51)
>>>>>>>>> at org.apache.flink.runtime.io
>>>>>>>>> .network.api.serialization.SpanningRecordSerializer.addRecord(SpanningRecordSerializer.java:76)
>>>>>>>>> at org.apache.flink.runtime.io
>>>>>>>>> .network.api.writer.RecordWriter.emit(RecordWriter.java:83)
>>>>>>>>> at
>>>>>>>>> org.apache.flink.runtime.operators.shipping.OutputCollector.collect(OutputCollector.java:65)
>>>>>>>>> at
>>>>>>>>> org.apache.flink.runtime.operators.chaining.ChainedMapDriver.collect(ChainedMapDriver.java:78)
>>>>>>>>> at
>>>>>>>>> org.apache.flink.runtime.operators.chaining.ChainedMapDriver.collect(ChainedMapDriver.java:78)
>>>>>>>>> at
>>>>>>>>> org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:177)
>>>>>>>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
>>>>>>>>> at java.lang.Thread.run(Thread.java:724)
>>>>>>>>>
>>>>>>>>> Could this be because the RowSerializer does not support
null
>>>>>>>>> values? (Similar to Flink-629
>>>>>>>>> <https://issues.apache.org/jira/browse/FLINK-629>
)
>>>>>>>>>
>>>>>>>>> Currently, to overcome this issue, we are ignoring all
the rows
>>>>>>>>> which may have null values. For example, we have a method
cleanData defined
>>>>>>>>> as,
>>>>>>>>>
>>>>>>>>> def cleanData(table:Table, relevantColumns:Seq[String]):Table
= {
>>>>>>>>>     val whereClause: String = relevantColumns.map{
>>>>>>>>>         cName=>
>>>>>>>>>             s"$cName.isNotNull"
>>>>>>>>>     }.mkString(" && ")
>>>>>>>>>
>>>>>>>>>     val result :Table =
>>>>>>>>> table.select(relevantColumns.mkString(",")).where(whereClause)
>>>>>>>>>     result
>>>>>>>>> }
>>>>>>>>>
>>>>>>>>> Before operating on any Table, we use this method and
then
>>>>>>>>> continue with task.
>>>>>>>>>
>>>>>>>>> Is this the right way to handle this? If not please let
me know
>>>>>>>>> how to go about it.
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> Thanks,
>>>>>>>>> Shiti
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>
>>>
>>

Mime
View raw message