SAMOA-16: Add an adapter for Apache Flink-Streaming #11

senorcarbone · 2015-03-02T11:55:29Z

This PR includes support for Apache Flink as an adapter. We haven't updated the documentation yet but we could perhaps do it in a follow up PR, since it seems you are already working on a docs redesign.

gdfm · 2015-03-03T08:37:00Z

Thanks. There seem to be some errors in the tests (Kryo serialization for what I can see).
Probably some Kryo version mismatch. Could you have a look at them?

senorcarbone · 2015-03-03T13:17:39Z

Sure, we are looking into it! Kryo version incompatibility seems to be trickier than we thought.

gdfm · 2015-03-03T14:28:20Z

If we need to update Kryo on our side, we'll be happy to do so.

senorcarbone · 2015-03-03T16:07:23Z

It seems that the earliest Kryo version that Flink is compatible with is v2.23.0. However, Storm tests seem to fail to initialise the serialiser for versions greater than 2.17. Judging from the changelog It looks like the current Apache Storm version (v.0.10.0) supports newer Kryo versions so maybe a first approach would be to try upgrading Storm first. What do you think?

gdfm · 2015-03-03T16:27:03Z

Makes sense, as we want to update the Storm dependency anyway.
However, it will for sure happen to have conflicting dependencies on different platforms, and we don't want to mix them. The tests should ensure this isolation, i.e., never mix two platforms, and the different profiles aim at that.

We have a single kryo.version variable to ensure that we don't use different Kryo versions around the codebase, but in this case it seems to me that it is necessary.
If Flink depends on 2.23, but we have it set on 2.17, then we should be able to override it in the samoa-flink module. Do you agree?

senorcarbone · 2015-03-03T18:59:48Z

Good point. I just overrided the kryo version in the flink build and tested it locally. Also, all module tests seem to be passing on travis.

arinto · 2015-03-10T06:18:59Z

I can't build this PR successfully on my local. Here's the error message:

[ERROR] Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.1:compile (default-compile) on project samoa-flink: Compilation failure: Compilation failure:
[ERROR] /Users/arinto/git/incubator-samoa/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkEntranceProcessingItem.java:[54,79] <anonymous com.yahoo.labs.flink.topology.impl.FlinkEntranceProcessingItem$1> is not abstract and does not override abstract method cancel() in org.apache.flink.streaming.api.function.source.SourceFunction
[ERROR] /Users/arinto/git/incubator-samoa/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkEntranceProcessingItem.java:[64,25] method does not override or implement a method from a supertype

Apparently there are API changes apache/flink@8436e9c due to FLINK-1625. Probably we can use stable version of Flink for this PR. What do you guys think?

senorcarbone · 2015-03-10T08:39:19Z

hey Arinto! Thanks for looking into it. I agree, let's stick to the last stable release (0.8.1) for now. We will commit some appropriate changes for it today. cheers

senorcarbone · 2015-03-11T15:11:47Z

We reviewed the changes after 0.8.1 and a lot of things have changed since then, many constructs such as the type extractors we are currently using will not be functioning in 0.8.1. Thus, we decided to just make a final patch for the first RC of 0.9.0 that is coming very soon, most probably end of next week. It looks like it is the best way to go :)

rmetzger · 2015-03-11T15:15:14Z

+1
I think we need to release some kind of 0.9-preview release soon.

gdfm · 2015-03-30T07:42:24Z

Hi,
if I understand correctly, you are waiting for Flink to release 0.9 before updating the PR, am I correct?

senorcarbone · 2015-03-30T08:28:41Z

yes, we are planning a 0.9.0-milestone release so we can use that as the first supported stable version for the PR

[Flink] added the missing license [flink] Changes to Flink integration to SAMOA [flink] changes in order to debug [flink] Change Utils class [flink] minor refactorings [flink] StreamExecutionEnvironment passed through the factory Removed .iml files from git Add algorithm for detecting circles in the topology

Added more checkes in the circleCanBeInitialised function. Added setters/getter change kryo version to pom file added debug printouts in FlinkProcessingItem class

…nitialisation

senorcarbone · 2015-05-04T09:20:33Z

Hey @gdfm @arinto @abifet . Thanks for the review so far. Let me know if you have any additional comments to look into. I think there is nothing critical left, we will keep supporting the flink adapter with subsequent updates such as new releases and improvements.

gdfm · 2015-05-05T08:12:02Z

Hi @senorcarbone,
Thanks, I will have another look at it tomorrow.
We also need another review from @abifet or @arinto.
And thanks for offering to keep supporting the adapter :)

abifet · 2015-05-06T01:08:32Z

Hi @senorcarbone,

This adapter for Apache Flink seems amazing! Thanks so much! I tested and reviewed, and found the same issues as @gdfm. The accuracy of the VHT was quite good, and it was very slow in my computer. Any thoughts about this? Do we need to tune any parameter in the conf file?

rmetzger · 2015-05-07T07:22:51Z

We have a open JIRA in Flink to start a "streaming only" mode, which starts up Flink in an streaming optimized mode.

For now, you can set the following configuration values in the conf/flink-conf.yaml to achieve a similar effect:

taskmanager.memory.fraction: This fraction describes the division between managed and unmanaged heap space. Flink batch is using managed memory for its operations, flink streaming uses unmanaged memory only. By default the value is set to 0.7, which leaves not much memory for streaming. I would recommend setting the value to 0.001 ;)
taskmanager.heap.mb: Thats an integer specifying the task manager's heap size. By default its set to 512. I suspect your machine has more memory.

senorcarbone · 2015-05-09T13:23:10Z

Hello again @gdfm and @abifet ,
I did a lot of cross-profiling between storm and flink, running the same VerticalHoeffdingTree task under different configurations during the last two days and I think the results are quite interesting.

It looks like the algorithm performance (and accuracy) depends heavily on the ingestion speed of the local statistics processors. The irony is that the greater the speed the slower the whole computation gets by time since more and more attribute events are sent to the local statistics processors with higher rate, the more updates the model aggregator gets back.

The average processing delay (in num of flatten instances processed by the aggregator between sending a process event and receiving the respective local statistics) is ~2k instances for Flink and around 400k instances for Storm. Also in Storm the aggregator continuously broadcasts ~100-200 attribute messages to local processors on average while Flink broadcasts ~2100 attribute messages due to the rate it gets results back I assume. These are collected locally on each component and there was no message duplication.
Since you worked on the algorithm, do you find this behavior reasonable?

senorcarbone · 2015-05-09T13:33:38Z

I also tried adding a 2sec sleep on the local flink processors to delay their results back and the algorithm finished in 12sec with less (57%) accuracy. Perhaps the model aggregator could be enhanced with some flow control logic to compensate between model updating rates and accuracy.

gdfm · 2015-05-10T13:55:30Z

Hi @senorcarbone,
Thanks for digging into the issue.
Interesting findings, though I still need to digest what they mean exactly.
We know that accuracy depends on speed, and the higher the speed the lower the accuracy. What I expect, however, is that the number of messages generated by the algorithm is independent of the speed (at least mostly). If this is not the case, there must be some wrong assumption I am making about the system.

senorcarbone · 2015-05-10T16:21:29Z

I think it has to do with the number of splits. If the model aggr. gets more local statistics during the time it processes the same amount of instances, it will have to split (exponentially?) more times and send more attributes in the same period. That is what I got from the experiments at least.

Maybe there should be a separate issue on VHT but you can keep me and @fobeligi in sync regarding it. @fobeligi is working on an experimental native implementation of VHT on Flink which we can share soon and she had to deal with similar issues. Also, feel free to let me know if you want me to try out more experiments and share results to speed up the process.

Regarding the PR, do you think we should look into something more?

gdfm · 2015-05-13T10:35:33Z

Indeed, I see what you mean. Given that the feedback loop in Flink is faster, the number of attempts to split should increase.
This is expected, but the number of such attempts is upper bounded by the ones tried on the Local engine, where there is no delay between request of the split criterion and response by the local statistics.

We already have some flow control to regulate the rate of ingestion in PrequentialEvaluation. I'll play a bit with it to see what happens.
When you put the 2 seconds delay in the Flink Processors, what happens (I guess) is that the whole data streams through a very rough, sub-optimal version of the tree. So it's very fast, but the precision drops considerably because of the artificial limit on the number of split attempts.

gdfm · 2015-05-18T09:03:14Z

samoa-instances/src/main/java/com/yahoo/labs/samoa/instances/SingleLabelInstance.java

+	public void setInstanceInformation(InstancesHeader instanceInformation) {
+		this.instanceInformation = instanceInformation;
+	}
+


Why do we need these changes for Flink?

these were test leftovers apparently, we do need them according to @fobeligi

gdfm · 2015-05-18T09:05:33Z

Hi,

I think we should address the issue with VHT in a separate PR.
I'd like to finalize this one.

@senorcarbone there are just a few issues outstanding.
There are a few files in samoa-api/build which should not be part of the patch.
Also, there is a question about the partitioning enum that I'd like to be answered.
Finally, there are some changes in samoa-instances that are not clear to me (I had missed them before).

Once these are fixed I'm +1

senorcarbone · 2015-05-18T15:31:42Z

Thanks @gdfm . I hope I addressed everything in the last commit and inline comments. Let me know if you want me to look into something more!

gdfm · 2015-05-25T14:06:28Z

Yes, the patch looks good to me.
+1

@abifet @arinto any comment?
I will merge the patch tomorrow otherwise.

abifet · 2015-05-26T00:35:54Z

+1

gdfm · 2015-05-26T08:35:54Z

Merged.
@senorcarbone thanks for the contribution!

rmetzger · 2015-05-26T08:44:43Z

Great to see this merged.

Is the Samoa website located in the gh-pages branch?
I would like to update the front page to mention Flink as well ;)

gdfm · 2015-05-26T08:47:49Z

Yes, the sources are in gh-pages and we use jekyll to build the website.
@rmetzger it would be great if you could add docs :)
Instructions here: https://cwiki.apache.org/confluence/display/SAMOA/SAMOA+Home

rmetzger · 2015-05-26T09:02:17Z

Thank you.
I've filed a JIRA: https://issues.apache.org/jira/browse/SAMOA-33
Lets see when I have time for this ;)

senorcarbone · 2015-05-29T13:35:09Z

Awesome, thanks @gdfm! Will you merge from local branch? Mind that it is missing some last commits. I will close the PR once everything is merged.

On a slightly related note, we ran some experiments today on the new stable streaming api that will be released soon that contains many good fixes and additions. The VHT experiment mentioned above now takes approximately around 80sec for 1m instances since filters are properly chained, making input ingestion much faster so that it catches up with local processors. You can find the branch with the patch here. Special thanks go to @gyfora for fixing chaining in iterations!

I will create a new PR once we have a stable release.

gdfm · 2015-05-29T16:02:26Z

Weird, I have already merged the PR and pushed back to Apache.
https://git-wip-us.apache.org/repos/asf?p=incubator-samoa.git

Something in the mirroring is not working, the PR should be closed automatically.

Thanks for the info, improving performances would be a great contribution!

senorcarbone changed the title ~~Flink integration~~ SAMOA-16: Add an adapter for Apache Flink-Streaming Mar 2, 2015

senorcarbone force-pushed the flink-integration branch from d472085 to d2570cb Compare March 31, 2015 14:57

senorcarbone and others added 16 commits April 20, 2015 21:38

[flink] Initial commit for Flink Integration

9615d52

[flink] pom fix and license headers added

930f7f5

[flink] Circle detection and circlular component initialization

d63a900

Added more checkes in the circleCanBeInitialised function. Added setters/getter change kryo version to pom file added debug printouts in FlinkProcessingItem class

[flink] minor refactorings added for debugging

08a05ec

[flink] SamoaTypeInfo and Serializer added.

9eddb4a

[flink] made a fix for grouped datastreams

cf62bfd

[flink] replaced output splitting with filters

427badd

[flink] made proper samoatype serialization

50fca38

[flink] Minor refactoring of code-change FlinkStream Ids getters/setters

9635940

[flink] typeinformation fix and rich functions for proper component i…

4c4df40

…nitialisation

[flink]updated circle initialization and detection

a5be987

[flink] Modify Samoa executable and Refactoring of Utils file

a6abd98

[flink] minor changes

cb110e8

[flink]Minor changes to execution parameters

dcf3047

[flink]Minor change to arguments

22e3385

gdfm mentioned this pull request May 5, 2015

SAMOA-27: Storm upgrade to the latest stable version (0.9.4) #25

Closed

[flink] set correct parallelism for filters

915fad3

gdfm reviewed May 18, 2015
View reviewed changes

senorcarbone force-pushed the flink-integration branch 5 times, most recently from 18d6d27 to 8c54b73 Compare May 18, 2015 14:25

[flink] cleanup of unused and generated files

493c4c9

senorcarbone force-pushed the flink-integration branch from 8c54b73 to 493c4c9 Compare May 18, 2015 15:05

asfgit closed this in 64ef7a9 Jun 3, 2015

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

SAMOA-16: Add an adapter for Apache Flink-Streaming #11

SAMOA-16: Add an adapter for Apache Flink-Streaming #11

senorcarbone commented Mar 2, 2015

gdfm commented Mar 3, 2015

senorcarbone commented Mar 3, 2015

gdfm commented Mar 3, 2015

senorcarbone commented Mar 3, 2015

gdfm commented Mar 3, 2015

senorcarbone commented Mar 3, 2015

arinto commented Mar 10, 2015

senorcarbone commented Mar 10, 2015

senorcarbone commented Mar 11, 2015

rmetzger commented Mar 11, 2015

gdfm commented Mar 30, 2015

senorcarbone commented Mar 30, 2015

senorcarbone commented May 4, 2015

gdfm commented May 5, 2015

abifet commented May 6, 2015

rmetzger commented May 7, 2015

senorcarbone commented May 9, 2015

senorcarbone commented May 9, 2015

gdfm commented May 10, 2015

senorcarbone commented May 10, 2015

gdfm commented May 13, 2015

gdfm May 18, 2015

senorcarbone May 18, 2015

gdfm commented May 18, 2015

senorcarbone commented May 18, 2015

gdfm commented May 25, 2015

abifet commented May 26, 2015

gdfm commented May 26, 2015

rmetzger commented May 26, 2015

gdfm commented May 26, 2015

rmetzger commented May 26, 2015

senorcarbone commented May 29, 2015

gdfm commented May 29, 2015

SAMOA-16: Add an adapter for Apache Flink-Streaming #11

SAMOA-16: Add an adapter for Apache Flink-Streaming #11

Conversation

senorcarbone commented Mar 2, 2015

gdfm commented Mar 3, 2015

senorcarbone commented Mar 3, 2015

gdfm commented Mar 3, 2015

senorcarbone commented Mar 3, 2015

gdfm commented Mar 3, 2015

senorcarbone commented Mar 3, 2015

arinto commented Mar 10, 2015

senorcarbone commented Mar 10, 2015

senorcarbone commented Mar 11, 2015

rmetzger commented Mar 11, 2015

gdfm commented Mar 30, 2015

senorcarbone commented Mar 30, 2015

senorcarbone commented May 4, 2015

gdfm commented May 5, 2015

abifet commented May 6, 2015

rmetzger commented May 7, 2015

senorcarbone commented May 9, 2015

senorcarbone commented May 9, 2015

gdfm commented May 10, 2015

senorcarbone commented May 10, 2015

gdfm commented May 13, 2015

gdfm May 18, 2015

Choose a reason for hiding this comment

senorcarbone May 18, 2015

Choose a reason for hiding this comment

gdfm commented May 18, 2015

senorcarbone commented May 18, 2015

gdfm commented May 25, 2015

abifet commented May 26, 2015

gdfm commented May 26, 2015

rmetzger commented May 26, 2015

gdfm commented May 26, 2015

rmetzger commented May 26, 2015

senorcarbone commented May 29, 2015

gdfm commented May 29, 2015