Mailing-List: contact commits-help@beam.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@beam.apache.org
Date: Thu, 20 Apr 2017 03:20:04 +0000 (UTC)
From: "tianyou (JIRA)" <jira@apache.org>
To: commits@beam.apache.org
Message-ID: <JIRA.13058491.1490251599000.1856.1492658404171@Atlassian.JIRA>
In-Reply-To: <JIRA.13058491.1490251599000@Atlassian.JIRA>
References: <JIRA.13058491.1490251599000@Atlassian.JIRA> <JIRA.13058491.1490251599994@jira-lw-us.apache.org>
Subject: [jira] [Issue Comment Deleted] (BEAM-1789) window can't not use in
 spark cluster module
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: quoted-printable
archived-at: Thu, 20 Apr 2017 03:20:09 -0000


     [ https://issues.apache.org/jira/browse/BEAM-1789?page=3Dcom.atlassian=
.jira.plugin.system.issuetabpanels:all-tabpanel ]

tianyou updated BEAM-1789:
--------------------------
    Comment: was deleted

(was: I change *strong*Window.Bound<KV<String,String>> fixWindow =3D Window=
.<KV<String,String>> into(FixedWindows.of(size));*strong*)

> window can't not use in spark cluster module
> --------------------------------------------
>
>                 Key: BEAM-1789
>                 URL: https://issues.apache.org/jira/browse/BEAM-1789
>             Project: Beam
>          Issue Type: Bug
>          Components: runner-spark
>            Reporter: tianyou
>            Assignee: Amit Sela
>
>  I user beam in spark cluster,The application is blow.
> SparkPipelineOptions options =3D PipelineOptionsFactory.as(SparkPipelineO=
ptions.class);
>         options.setRunner(SparkRunner.class);
>         options.setEnableSparkMetricSinks(false);
>         options.setStreaming(true);
>         options.setSparkMaster("spark://10.100.124.205:6066");
>         options.setAppName("Beam App Spark"+new Random().nextFloat());
>         options.setJobName("Beam Job Spark"+new Random().nextFloat());
>         System.out.println("App Name:"+options.getAppName());
>         System.out.println("Job Name:"+options.getJobName());
>         options.setMaxRecordsPerBatch(100000L);
>        =20
> //      PipelineOptions options =3D PipelineOptionsFactory.create();
>         Pipeline p =3D Pipeline.create(options);
>        =20
> //      Duration size =3D Duration.standardMinutes(4);
>         long duration =3D 60;
>         if(args!=3Dnull && args.length=3D=3D1){
>             duration =3D Integer.valueOf(args[0]);
>         }
>         Duration size =3D Duration.standardSeconds(duration);
>         System.out.println("=E6=97=B6=E9=97=B4=E7=AA=97=E5=8F=A3=E4=B8=BA=
:["+duration+"]=E7=A7=92");
>         Window.Bound<KV<String,String>> fixWindow =3D Window.<KV<String,S=
tring>> into(
>             FixedWindows.of(size)
>         );
>        =20
>         String kafkaAddress =3D "10.100.124.208:9093";
> //      String kafkaAddress =3D "192.168.100.212:9092";
>        =20
>         Map<String, Object> kfConsunmerConf =3D new HashMap<String, Objec=
t>();
>         kfConsunmerConf.put("auto.offset.reset", "latest");
>         PCollection<String> kafkaJsonPc =3D p.apply(KafkaIO.<String, Stri=
ng> read()
>             .withBootstrapServers(kafkaAddress)
>             .withTopics(ImmutableList.of("wypxx1"))
>             .withKeyCoder(StringUtf8Coder.of())=20
>             .withValueCoder(StringUtf8Coder.of())
>             .updateConsumerProperties(kfConsunmerConf)
>             .withoutMetadata()=20
>         ).apply(Values.<String> create());
>        =20
>        =20
>         PCollection<KV<String,String>> totalPc =3D kafkaJsonPc.apply(
>                 "count line",
>                 ParDo.of(new DoFn<String,KV<String,String>>() {
>                     @ProcessElement
>                       public void processElement(ProcessContext c) {
>                         String line =3D c.element();
>                         Instant is =3D c.timestamp();
>                         if(line.length()>2)
>                           line =3D line.substring(0,2);
>                         System.out.println(line + " " +  is.toString());
>                         c.output(KV.of(line, line));
>                       }
>                  })
>             );
>            =20
>        =20
>         PCollection<KV<String, Iterable<String>>> itPc =3D totalPc.apply(=
fixWindow).apply(
>                 "group by appKey",
>                 GroupByKey.<String, String>create()
>             );
>           itPc.apply(ParDo.of(new DoFn<KV<String, Iterable<String>>, Void=
>() {
>                 @ProcessElement
>                 public void processElement(ProcessContext c) {
>                     KV<String, Iterable<String>> keyIt =3D c.element();
>                     String key =3D keyIt.getKey();
>                     Iterable<String> itb =3D keyIt.getValue();
>                     Iterator<String> it =3D itb.iterator();
>                     StringBuilder sb =3D new StringBuilder();
>                     sb.append(key).append(":[");
>                     while(it.hasNext()){
>                         sb.append(it.next()).append(",");
>                     }
>                     String str =3D sb.toString();
>                     str =3D str.substring(0,str.length() -1) + "]";
>                     System.out.println(str);
>                     String filePath =3D "/data/wyp/sparktest.txt";
>                     String line =3D "word-->["+key+"]total count=3D"+str+=
"--->time+"+c.timestamp().toString();
>                     System.out.println("writefile----->"+line);
>                     FileUtil.write(filePath, line, true, true);
>                 }
>                =20
>              }));
>          =20
>             p.run().waitUntilFinish();
> When I user submit application to spark cluster.In spark UI,I can see log=
 of  totalPc PCollection  of. after one miniter but I can.t see log of itPc=
 PCollection.
> I use local mode spark,It work well.
> Please help me to resovle this proplems.Thanks!


--
This message was sent by Atlassian JIRA
(v6.3.15#6346)