Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@flink.apache.org
Content-type: text/plain; charset=utf-8
MIME-version: 1.0 (Mac OS X Mail 8.2 \(2098\))
Subject: Re: Apache Flink transactions
From: Chiwan Park <chiwanpark@icloud.com>
In-reply-to: <1433463751772-1491.post@n4.nabble.com>
Date: Fri, 05 Jun 2015 11:55:07 +0900
Content-transfer-encoding: quoted-printable
Message-id: <F8220DEE-E58A-40B9-83DD-B7759E98A534@icloud.com>
References: <6C25AA06-E4A6-4647-8642-30744E2BDFAE@icloud.com>
 <1433463751772-1491.post@n4.nabble.com>
To: user@flink.apache.org

Basically Flink uses Data Model in functional programming model. All =
DataSet is immutable. This means we cannot modify DataSet but =E3=85=90onl=
y can create new DataSet with modification. Update, delete query are =
represented as writing filtered DataSet.
Following scala sample shows select, insert, update, and remove query in =
Flink. (I=E2=80=99m not sure this is best practice.)

case class MyType(id: Int, value1: String, value2: String)

// load data (you can use readCsvFile, or something else.)
val data =3D env.fromElements(MyType(0, =E2=80=9Ctest=E2=80=9D, =
=E2=80=9Ctest2=E2=80=9D), MyType(1, =E2=80=9Chello=E2=80=9D, =
=E2=80=9Cflink=E2=80=9D), MyType(2, =E2=80=9Cflink=E2=80=9D, =
=E2=80=9Cgood=E2=80=9D))

// selecting
// same as SELECT * FROM data WHERE id =3D 1
val selectedData1 =3D data.filter(_.id =3D=3D 1)
// same as SELECT value1 FROM data WHERE id =3D 1
val selectedData2 =3D data.filter(_.id =3D=3D 1).map(_.value1)

// removing is same as selecting such as following
// same as DELETE FROM data WHERE id =3D 1, but DataSet data is not =
changed. the result is removedData
val removedData =3D data.filter(_.id !=3D 1)

// inserting
// same as INSERT INTO data (id, value1, value2) VALUES (3, =E2=80=9Cnew=E2=
=80=9D, =E2=80=9Cdata=E2=80=9D)
val newData =3D env.fromElements(MyType(3, =E2=80=9Cnew=E2=80=9D, =
=E2=80=9Cdata=E2=80=9D))
val insertedData =3D data.union(newData)=20

// updating
// UPDATE data SET value1 =3D =E2=80=9Cupdated=E2=80=9D, value2 =3D =
=E2=80=9Cdata=E2=80=9D WHERE id =3D 1, but DataSet data is not changed.
val updatedData =3D data.map { x =3D> if (x.id =3D=3D 1) MyType(x.id, =
=E2=80=9Cupdated=E2=80=9D, =E2=80=9Cdata=E2=80=9D) else x }=20

Regards,
Chiwan Park

> On Jun 5, 2015, at 9:22 AM, hawin <hawin.jiang@gmail.com> wrote:
>=20
> Hi  Chiwan
>=20
> Thanks for your information.  I knew Flink is not DBMS. I want to know =
what
> is the flink way to select, insert, update and delete data on HDFS.
>=20
>=20
> @Till
> Maybe union is a way to insert data. But I think it will cost some
> performance issue.
>=20
>=20
> @Stephan
> Thanks for your suggestion.  I have checked apache flink roadmap.  SQL =
on
> flink will be released on Q3/Q4 2015. Will it support insertion, =
deletion
> and update data on HDFS?
> You guys already provided a nice example for selecting data on HDFS.  =
Such
> as: TPCHQuery10 and TPCHQuery3.=20
> Do you have other examples for inserting, updating and removing data =
on HDFS
> by Apache flink
>=20
> Thanks=20
>=20
>=20
>=20
>=20
> --
> View this message in context: =
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Re-Apa=
che-Flink-transactions-tp1457p1491.html
> Sent from the Apache Flink User Mailing List archive. mailing list =
archive at Nabble.com.