ddf-project / ddf-flink Goto Github PK

View Code? Open in Web Editor NEW

3.0 17.0 10.0 12.1 MB

DDF with Flink Implementation

License: Apache License 2.0

Shell 14.01% Java 5.25% Scala 80.31% Batchfile 0.43%

ddf-flink's Introduction

DDF with Flink

This project depends on DDF and uses Apache Flink engine.

DDF

Distributed DataFrame: Productivity = Power x Simplicity For Big Data Scientists & Engineers

Getting Started

This project depends on DDF core and requires its installation to run. To get DDF core, clone DDF repo and checkout the master branch.

$ git clone [email protected]:ddf-project/DDF.git
$ cd DDF

DDF can be installed by,

$ sbt publishLocal

Installing ddf-with-flink can be done by

$ git clone [email protected]:ddf-project/ddf-flink.git
$ cd ddf-with-flink
$ bin/run-once.sh
$ mvn package install -DskipTests

Running tests

Tests can be run either through SBT or Maven,

$ sbt test
$ mvn test

//running a single test

$ sbt "testOnly *FlinkDDFManagerSpec*"

$ mvn test -Dsuites='io.ddf.flink.FlinkDDFManagerSpec'

Starting `ddf-shell` with `flink` engine

Execute the following only after installing ddf-with-flink

$ sbt package
$ bin/ddf-shell

SBT package is required since it generates the lib_managed which is required for running the scripts.

Running the example,

$ sbt package
$ bin/run-flink-example io.ddf.flink.examples.FlinkDDFExample

SBT package is required since it generates the lib_managed which is required for running the scripts.

####Todo

Test the ML method getConfusionMatrix
Implement transformPython and flattenDDF for TransformationHandler and also test the R functions.
Implement the methods r2score, residuals, roc and rmse for MLMetricsSupporter

ddf-flink's People

Contributors

Stargazers

Watchers

Forkers

shiti milliondreams tuplejump codeaudit reserach vrosnet jonyboy2000

ddf-flink's Issues

Rename scala test class: Spec.scala to Suite.scala

Document approach for ML Support.

io.ddf.flink rather than io.flink.ddf

Sorry about that, this was just a silly package naming error in io.spark.ddf, which itself should have been io.ddf.spark. See ddf-project/DDF#122

Document DDF shell integration.

Come up with approach for Flink integration.

Identify the test scenarios

We should review the Spark implementation test cases and document and write test specs for our Flink implementation

Derive schema from MRQL result

Infer the schema of a sql2ddf command MRQL Top Level query type.

Add flink project into DDF repo and integrate with build system.

Use schema derived from MRQL result to make a Flink DataSet

Implement Flink DDF with Flink Table support

Upgrade the Flink version to 0.9.1 and implement FlinkDDF using a "Table" based representation

Write new ISupportML and IModel to work with FlinkML pipelines

Implement Representation Handlers For ML

We need to add representation handlers to convert the DDF to relevant formats as accepted by the ML Algorithms.

Some lines in airline.csv are being ignored.

Setup email notification in jenkins to notify when flink tests fail

README for Flink project for new devs

Document basic Spark Support for understanding DDF internals

Document SQL syntax supported by the custom Flink SQL Parser.

JoinHandler. Research how left semi and full outer join can be implemented.

Restrict third-party application access to GitHub repo

Hi @tuplejump/owners, apparently GitHub defaults to open access via OAUTH, unless you specifically reconfigure it. Please go to https://github.com/organizations/tuplejump/settings/oauth_application_policy, and click on "Set up application access restrictions", then "Restrict third-party application access".

I ran into this when authorizing spark-packages to connect to ddf-project, then noticed that it proceeds to propose to grant access to all the repos I have access to, which is not a good idea.

Cheers,

https://www.dropbox.com/s/pk6xe0vxfhs06ni/Screenshot%202015-06-17%2000.51.56.png?dl=0

getSchema
getColumns
getNumColumns

SQLHandler does not support complex query

The following query throws an illegal args exception

select * from airlineNA where ( (case when Year is null then 1 else 0 end) + (case when Month is null then 1 else 0 end) + (case when DayofMonth is null then 1 else 0 end) + (case when DayOfWeek is null then 1 else 0 end) + (case when DepTime is null then 1 else 0 end) + (case when CRSDepTime is null then 1 else 0 end) + (case when ArrTime is null then 1 else 0 end) + (case when CRSArrTime is null then 1 else 0 end) + (case when UniqueCarrier is null then 1 else 0 end) + (case when FlightNum is null then 1 else 0 end) + (case when TailNum is null then 1 else 0 end) + (case when ActualElapsedTime is null then 1 else 0 end) + (case when CRSElapsedTime is null then 1 else 0 end) + (case when AirTime is null then 1 else 0 end) + (case when ArrDelay is null then 1 else 0 end) + (case when DepDelay is null then 1 else 0 end) + (case when Origin is null then 1 else 0 end) + (case when Dest is null then 1 else 0 end) + (case when Distance is null then 1 else 0 end) + (case when TaxiIn is null then 1 else 0 end) + (case when TaxiOut is null then 1 else 0 end) + (case when Cancelled is null then 1 else 0 end) + (case when CancellationCode is null then 1 else 0 end) + (case when Diverted is null then 1 else 0 end) + (case when CarrierDelay is null then 1 else 0 end) + (case when WeatherDelay is null then 1 else 0 end) + (case when NASDelay is null then 1 else 0 end) + (case when SecurityDelay is null then 1 else 0 end) + (case when LateAircraftDelay is null then 1 else 0 end) )< 1

The error message is

 java.lang.IllegalArgumentException: Cannot parse [select * from airlineNA where ( (case when Year is null then 1 else 0 end) + (case when Month is null then 1 else 0 end) + (case when DayofMonth is null then 1 else 0 end) + (case when DayOfWeek is null then 1 else 0 end) + (case when DepTime is null then 1 else 0 end) + (case when CRSDepTime is null then 1 else 0 end) + (case when ArrTime is null then 1 else 0 end) + (case when CRSArrTime is null then 1 else 0 end) + (case when UniqueCarrier is null then 1 else 0 end) + (case when FlightNum is null then 1 else 0 end) + (case when TailNum is null then 1 else 0 end) + (case when ActualElapsedTime is null then 1 else 0 end) + (case when CRSElapsedTime is null then 1 else 0 end) + (case when AirTime is null then 1 else 0 end) + (case when ArrDelay is null then 1 else 0 end) + (case when DepDelay is null then 1 else 0 end) + (case when Origin is null then 1 else 0 end) + (case when Dest is null then 1 else 0 end) + (case when Distance is null then 1 else 0 end) + (case when TaxiIn is null then 1 else 0 end) + (case when TaxiOut is null then 1 else 0 end) + (case when Cancelled is null then 1 else 0 end) + (case when CancellationCode is null then 1 else 0 end) + (case when Diverted is null then 1 else 0 end) + (case when CarrierDelay is null then 1 else 0 end) + (case when WeatherDelay is null then 1 else 0 end) + (case when NASDelay is null then 1 else 0 end) + (case when SecurityDelay is null then 1 else 0 end) + (case when LateAircraftDelay is null then 1 else 0 end) )< 1] because `)' expected but `w' found
[info]   at io.ddf.flink.content.SqlSupport$TableDdlParser.parse(SqlSupport.scala:302)
[info]   at io.ddf.flink.etl.SqlHandler.parse(SqlHandler.scala:26)
[info]   at io.ddf.flink.etl.SqlHandler.sql2ddf(SqlHandler.scala:29)
[info]   at io.ddf.flink.etl.SqlHandler.sql2ddf(SqlHandler.scala:154)

Find out how table environment can be used to reflect the schema.
Integrate MRQL Support as FlinkSQLSupport

Load table should add it to flink table environment.