tianyou created BEAM-1789: ----------------------------- Summary: 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 = PipelineOptionsFactory.as(SparkPipelineOptions.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); // PipelineOptions options = PipelineOptionsFactory.create(); Pipeline p = Pipeline.create(options); // Duration size = Duration.standardMinutes(4); long duration = 60; if(args!=null && args.length==1){ duration = Integer.valueOf(args[0]); } Duration size = Duration.standardSeconds(duration); System.out.println("时间窗口为:["+duration+"]秒"); Window.Bound<KV<String,String>> fixWindow = Window.<KV<String,String>> into( FixedWindows.of(size) ); String kafkaAddress = "10.100.124.208:9093"; // String kafkaAddress = "192.168.100.212:9092"; Map<String, Object> kfConsunmerConf = new HashMap<String, Object>(); kfConsunmerConf.put("auto.offset.reset", "latest"); PCollection<String> kafkaJsonPc = p.apply(KafkaIO.<String, String> read() .withBootstrapServers(kafkaAddress) .withTopics(ImmutableList.of("wypxx1")) .withKeyCoder(StringUtf8Coder.of()) .withValueCoder(StringUtf8Coder.of()) .updateConsumerProperties(kfConsunmerConf) .withoutMetadata() ).apply(Values.<String> create()); PCollection<KV<String,String>> totalPc = kafkaJsonPc.apply( "count line", ParDo.of(new DoFn<String,KV<String,String>>() { @ProcessElement public void processElement(ProcessContext c) { String line = c.element(); Instant is = c.timestamp(); if(line.length()>2) line = line.substring(0,2); System.out.println(line + " " + is.toString()); c.output(KV.of(line, line)); } }) ); PCollection<KV<String, Iterable<String>>> itPc = 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 = c.element(); String key = keyIt.getKey(); Iterable<String> itb = keyIt.getValue(); Iterator<String> it = itb.iterator(); StringBuilder sb = new StringBuilder(); sb.append(key).append(":["); while(it.hasNext()){ sb.append(it.next()).append(","); } String str = sb.toString(); str = str.substring(0,str.length() -1) + "]"; System.out.println(str); String filePath = "/data/wyp/sparktest.txt"; String line = "word-->["+key+"]total count="+str+"--->time+"+c.timestamp().toString(); System.out.println("writefile----->"+line); FileUtil.write(filePath, line, true, true); } })); 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)