Mailing-List: contact commits-help@beam.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@beam.apache.org
Date: Fri, 17 Nov 2017 17:05:00 +0000 (UTC)
From: "Luke Cwik (JIRA)" <jira@apache.org>
To: commits@beam.apache.org
Message-ID: <JIRA.13119124.1510897600000.268103.1510938300364@Atlassian.JIRA>
In-Reply-To: <JIRA.13119124.1510897600000@Atlassian.JIRA>
References: <JIRA.13119124.1510897600000@Atlassian.JIRA> <JIRA.13119124.1510897600133@jira-lw-us.apache.org>
Subject: [jira] [Commented] (BEAM-3210) The problem about the use of
 waitUntilFinish() in DirectRunner
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: quoted-printable
archived-at: Fri, 17 Nov 2017 17:05:06 -0000


    [ https://issues.apache.org/jira/browse/BEAM-3210?page=3Dcom.atlassian.=
jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=3D16257=
227#comment-16257227 ]=20

Luke Cwik commented on BEAM-3210:
---------------------------------

Your using a non-thread safe data structure and not synchronizing reads/wri=
tes.

The DirectRunner processes elements in parallel using multiple threads whic=
h is why the elements in your list are appearing out of order.

Consider using a thread safe data structure of synchronizing your list usin=
g [Collections#synchronizedList](https://docs.oracle.com/javase/7/docs/api/=
java/util/Collections.html#synchronizedList(java.util.List))
, either using a thread safe data structure or synchronizing your reads and=
 writes to you


> The problem about the use of waitUntilFinish() in DirectRunner
> --------------------------------------------------------------
>
>                 Key: BEAM-3210
>                 URL: https://issues.apache.org/jira/browse/BEAM-3210
>             Project: Beam
>          Issue Type: Bug
>          Components: runner-direct
>    Affects Versions: 2.1.0
>         Environment: Ubuntn 14.04.3 LTS
> JDK 1.8
> Beam 2.1.0
> Maven 3.5.0
>            Reporter: Rick Lin
>            Assignee: Thomas Groh
>             Fix For: 2.1.0
>
>
> Dear sir,
> The description of waitUntilFinish() is "waits until the pipeline finishe=
s and returns the final status."
> In my project, a static variable is used to record a PCollection context,=
 where the static variable is a data list type.
> For this, I considered the "p.run().waitUntilFinish()"  to wait until the=
 pipeline finishes to avoid the loss of record in the data list.
> Unfortunately, there is a problem that the data list{color:#d04437} *some=
times* {color}may record the "null" value instead of the realistic value
> In order to clearly explain, i provide my java code in the following.
> {color:#14892c}"import java.io.IOException;
> import java.util.ArrayList;
> import org.apache.beam.sdk.Pipeline;
> import org.apache.beam.sdk.options.PipelineOptions;
> import org.apache.beam.sdk.options.PipelineOptionsFactory;
> import org.apache.beam.sdk.transforms.Create;
> import org.apache.beam.sdk.transforms.DoFn;
> import org.apache.beam.sdk.transforms.Mean;
> import org.apache.beam.sdk.transforms.ParDo;
> import org.apache.beam.sdk.transforms.DoFn.ProcessContext;
> import org.apache.beam.sdk.transforms.DoFn.ProcessElement;
> public class BeamTestStatic extends Thread {
> =E3=80=80=E3=80=80public static ArrayList<Double> myList =3D new ArrayLis=
t<Double>();
> =E3=80=80=E3=80=80public static class StaticTest extends DoFn<Double, Voi=
d> {
> =E3=80=80=E3=80=80=E3=80=80=E3=80=80@ProcessElement=09=09=20
> =E3=80=80=E3=80=80=E3=80=80=09public void test(ProcessContext c) {
> =09=09=E3=80=80=E3=80=80myList.add(c.element());=09
> =E3=80=80=E3=80=80=E3=80=80=09}=09=09
> =E3=80=80=E3=80=80}
> =E3=80=80public static void main(String[] args) throws IOException {=09=
=09=09
> =09StaticTest testa=3Dnew StaticTest();
> =09PipelineOptions options =3D PipelineOptionsFactory.create();
> =09Pipeline p =3D Pipeline.create(options);
> =E3=80=80=E3=80=80PCollection<Double> data=3Dp.apply("Rawdata",
> Create.of(1.0,2.0,3.0,4.0,5.0,6.0,7.0,8.0,9.0,10.0,));
> =09PCollection<Void> listtest=3D data.apply(ParDo.of(testa));
> =E3=80=80=E3=80=80p.run().waitUntilFinish();
> =E3=80=80=E3=80=80System.out.println("mylist_size_a=3D"+myList.size());
> =09=09=20
>         for (int i =3D 0; i < myList.size(); i++) {
>         =09System.out.println("mylist_data=3D"+myList.get(i));
>         }
> "{color}
> In addition, the result of my code is:
> {color:#205081}"mylist_size_a=3D10
> mylist_data=3Dnull
> mylist_data=3D4.0
> mylist_data=3D5.0
> mylist_data=3D9.0
> mylist_data=3D6.0
> mylist_data=3D1.0
> mylist_data=3D7.0
> mylist_data=3D8.0
> mylist_data=3D10.0
> mylist_data=3D3.0"{color}
> If you have any further information, I am glad to be informed.
> Thanks
> Rick


--
This message was sent by Atlassian JIRA
(v6.4.14#64029)