org.apache.beam.sdk.transforms.windowing.Window.into()方法的使用及代码示例

x33g5p2x  于2022-02-03 转载在 其他  
字(10.1k)|赞(0)|评价(0)|浏览(165)

本文整理了Java中org.apache.beam.sdk.transforms.windowing.Window.into()方法的一些代码示例，展示了Window.into()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Window.into()方法的具体详情如下：
包路径：org.apache.beam.sdk.transforms.windowing.Window
类名称：Window
方法名：into

Window.into介绍

[英]Creates a Window PTransform that uses the given WindowFn to window the data.

The resulting PTransform's types have been bound, with both the input and output being a PCollection, inferred from the types of the argument WindowFn. It is ready to be applied, or further properties can be set on it first.
[中]创建一个窗口PTransform，该窗口PTransform使用给定的窗口fn打开数据窗口。
结果PTransform的类型已经绑定，输入和输出都是PCollection，由参数WindowFn的类型推断出来。它已准备好应用，或者可以先在其上设置进一步的属性。

代码示例

代码示例来源：origin: takidau/streamingbook

@Override
public PCollection<String> expand(PCollection<KV<String, Integer>> input) {
  return input
    .apply(Window.<KV<String, Integer>>into(FixedWindows.of(TWO_MINUTES)))
    .apply(Sum.integersPerKey())
    .apply(ParDo.of(new FormatAsStrings()));
}

代码示例来源：origin: org.apache.beam/beam-examples-java

@Override
 public PCollection<KV<String, Long>> expand(PCollection<String> actions) {
  return actions
    .apply(Window.into(Sessions.withGapDuration(Duration.standardHours(1))))
    .apply(Count.perElement());
 }
}

代码示例来源：origin: org.apache.beam/beam-sdks-java-core

@Test
@Category(NeedsRunner.class)
public void testSampleAnyEmpty() {
 PCollection<Integer> input = pipeline.apply(Create.empty(BigEndianIntegerCoder.of()));
 PCollection<Integer> output =
   input
     .apply(Window.into(FixedWindows.of(Duration.standardSeconds(3))))
     .apply(Sample.any(2));
 PAssert.that(output).satisfies(new VerifyCorrectSample<>(0, EMPTY));
 pipeline.run();
}

代码示例来源：origin: org.apache.beam/beam-sdks-java-core

@Test
public void testNonDeterministicWindowCoder() throws NonDeterministicException {
 FixedWindows mockWindowFn = Mockito.mock(FixedWindows.class);
 @SuppressWarnings({"unchecked", "rawtypes"})
 Class<Coder<IntervalWindow>> coderClazz = (Class) Coder.class;
 Coder<IntervalWindow> mockCoder = Mockito.mock(coderClazz);
 when(mockWindowFn.windowCoder()).thenReturn(mockCoder);
 NonDeterministicException toBeThrown =
   new NonDeterministicException(mockCoder, "Its just not deterministic.");
 Mockito.doThrow(toBeThrown).when(mockCoder).verifyDeterministic();
 thrown.expect(IllegalArgumentException.class);
 thrown.expectCause(Matchers.sameInstance(toBeThrown));
 thrown.expectMessage("Window coders must be deterministic");
 Window.into(mockWindowFn);
}

代码示例来源：origin: org.apache.beam/beam-sdks-java-core

/** Test a WriteFiles with a windowed PCollection. */
@Test
@Category(NeedsRunner.class)
public void testWriteWindowed() throws IOException {
 List<String> inputs =
   Arrays.asList(
     "Critical canary",
     "Apprehensive eagle",
     "Intimidating pigeon",
     "Pedantic gull",
     "Frisky finch");
 runWrite(
   inputs,
   new WindowAndReshuffle<>(Window.into(FixedWindows.of(Duration.millis(2)))),
   getBaseOutputFilename(),
   WriteFiles.to(makeSimpleSink()));
}

代码示例来源：origin: org.apache.beam/beam-sdks-java-extensions-sql

@Test
public void testUnsupportedGlobalWindowWithDefaultTrigger() {
 exceptions.expect(UnsupportedOperationException.class);
 pipeline.enableAbandonedNodeEnforcement(false);
 PCollection<Row> input =
   unboundedInput1.apply(
     "unboundedInput1.globalWindow",
     Window.<Row>into(new GlobalWindows()).triggering(DefaultTrigger.of()));
 String sql = "SELECT f_int2, COUNT(*) AS `size` FROM PCOLLECTION GROUP BY f_int2";
 input.apply("testUnsupportedGlobalWindows", SqlTransform.query(sql));
}

代码示例来源：origin: org.apache.beam/beam-sdks-java-core

@Test
public void testWindowGetName() {
 assertEquals(
   "Window.Into()",
   Window.<String>into(FixedWindows.of(Duration.standardMinutes(10))).getName());
}

代码示例来源：origin: org.apache.beam/beam-sdks-java-core

/** Test a WriteFiles with sessions. */
@Test
@Category(NeedsRunner.class)
public void testWriteWithSessions() throws IOException {
 List<String> inputs =
   Arrays.asList(
     "Critical canary",
     "Apprehensive eagle",
     "Intimidating pigeon",
     "Pedantic gull",
     "Frisky finch");
 runWrite(
   inputs,
   new WindowAndReshuffle<>(Window.into(Sessions.withGapDuration(Duration.millis(1)))),
   getBaseOutputFilename(),
   WriteFiles.to(makeSimpleSink()));
}

代码示例来源：origin: org.apache.beam/beam-runners-core-java

@Before
public void setup() {
 MockitoAnnotations.initMocks(this);
 PCollection<Integer> created = p.apply(Create.of(1, 2, 3));
 singletonView =
   created
     .apply(Window.into(new IdentitySideInputWindowFn()))
     .apply(Sum.integersGlobally().asSingletonView());
 underlying = new TestDoFnRunner<>();
}

代码示例来源：origin: org.apache.beam/beam-sdks-java-core

@Test
@Category({ValidatesRunner.class, DataflowPortabilityApiUnsupported.class})
public void testWindowedCombineEmpty() {
 PCollection<Double> mean =
   pipeline
     .apply(Create.empty(BigEndianIntegerCoder.of()))
     .apply(Window.into(FixedWindows.of(Duration.millis(1))))
     .apply(Combine.globally(new MeanInts()).withoutDefaults());
 PAssert.that(mean).empty();
 pipeline.run();
}

代码示例来源：origin: org.apache.beam/beam-runners-core-construction-java

@Test
public void classEqualToDoesNotMatchUnrelatedClass() {
 PTransformMatcher matcher = PTransformMatchers.classEqualTo(ParDo.SingleOutput.class);
 AppliedPTransform<?, ?, ?> application =
   getAppliedTransform(Window.<KV<String, Integer>>into(new GlobalWindows()));
 assertThat(matcher.matches(application), is(false));
}

代码示例来源：origin: org.apache.beam/beam-sdks-java-core

@Test
@Category(NeedsRunner.class)
public void testWriteSpilling() throws IOException {
 List<String> inputs = Lists.newArrayList();
 for (int i = 0; i < 100; ++i) {
  inputs.add("mambo_number_" + i);
 }
 runWrite(
   inputs,
   Window.into(FixedWindows.of(Duration.millis(2))),
   getBaseOutputFilename(),
   WriteFiles.to(makeSimpleSink())
     .withMaxNumWritersPerBundle(2)
     .withWindowedWrites()
     .withNumShards(1));
}

代码示例来源：origin: org.apache.beam/beam-runners-google-cloud-dataflow-java

@Test
public void testInvalidWindowsService() {
 Pipeline p = createTestServiceRunner();
 List<KV<String, Integer>> ungroupedPairs = Arrays.asList();
 PCollection<KV<String, Integer>> input =
   p.apply(
       Create.of(ungroupedPairs)
         .withCoder(KvCoder.of(StringUtf8Coder.of(), BigEndianIntegerCoder.of())))
     .apply(Window.into(Sessions.withGapDuration(Duration.standardMinutes(1))));
 thrown.expect(IllegalStateException.class);
 thrown.expectMessage("GroupByKey must have a valid Window merge function");
 input.apply("GroupByKey", GroupByKey.create()).apply("GroupByKeyAgain", GroupByKey.create());
}

代码示例来源：origin: org.apache.beam/beam-sdks-java-core

@Test
public void testAssignDisplayDataUnchanged() {
 FixedWindows windowFn = FixedWindows.of(Duration.standardHours(5));
 Window<Object> original = Window.into(windowFn);
 WindowingStrategy<?, ?> updated = WindowingStrategy.globalDefault().withWindowFn(windowFn);
 DisplayData displayData = DisplayData.from(new Window.Assign<>(original, updated));
 assertThat(displayData, hasDisplayItem("windowFn", windowFn.getClass()));
 assertThat(displayData, includesDisplayDataFor("windowFn", windowFn));
 assertThat(displayData, not(hasDisplayItem("trigger")));
 assertThat(displayData, not(hasDisplayItem("accumulationMode")));
 assertThat(displayData, not(hasDisplayItem("allowedLateness")));
 assertThat(displayData, not(hasDisplayItem("closingBehavior")));
 assertThat(displayData, not(hasDisplayItem("timestampCombiner")));
}

代码示例来源：origin: takidau/streamingbook

@Override
  public PCollection<String> expand(PCollection<KV<String, Integer>> input) {
    return input
      .apply(Window.<KV<String, Integer>>into(FixedWindows.of(TWO_MINUTES)))
      .apply(Sum.integersPerKey())
      .apply(ParDo.of(new FormatAsStrings()));
  }
}

代码示例来源：origin: org.apache.beam/beam-runners-direct-java

@Override
 public PCollectionView<Integer> expand(PCollection<T> records) {
  return records
    .apply(Window.into(new GlobalWindows()))
    .apply("CountRecords", Count.globally())
    .apply("GenerateShardCount", ParDo.of(new CalculateShardsFn()))
    .apply(View.asSingleton());
 }
}

代码示例来源：origin: org.apache.beam/beam-sdks-java-core

@Test
public void testIncompatibleWindowFnPropagationFailure() {
 p.enableAbandonedNodeEnforcement(false);
 PCollection<String> input1 =
   p.apply("CreateInput1", Create.of("Input1"))
     .apply("Window1", Window.into(FixedWindows.of(Duration.standardMinutes(1))));
 PCollection<String> input2 =
   p.apply("CreateInput2", Create.of("Input2"))
     .apply("Window2", Window.into(FixedWindows.of(Duration.standardMinutes(2))));
 try {
  PCollectionList.of(input1).and(input2).apply(Flatten.pCollections());
  Assert.fail("Exception should have been thrown");
 } catch (IllegalStateException e) {
  Assert.assertTrue(
    e.getMessage().startsWith("Inputs to Flatten had incompatible window windowFns"));
 }
}

代码示例来源：origin: org.apache.beam/beam-sdks-java-core

@Test
public void testDisplayDataExcludesDefaults() {
 Window<?> window =
   Window.into(new GlobalWindows())
     .triggering(DefaultTrigger.of())
     .withAllowedLateness(Duration.millis(BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis()));
 DisplayData data = DisplayData.from(window);
 assertThat(data, not(hasDisplayItem("trigger")));
 assertThat(data, not(hasDisplayItem("allowedLateness")));
}

代码示例来源：origin: Talend/components

/**
 * Applies a window to the input collection if one hasn't already been specified.
 *
 * @return the input collection if it already has been windowed, otherwise a the same collection inside a default
 * window.
 */
public static <T> PCollection<T> ofDefaultWindow(PCollection<T> in) {
  if (in.getWindowingStrategy() != WindowingStrategy.globalDefault() && in.getWindowingStrategy() != null)
    return in;
  return in.apply("ApplyDefaultWindow", Window.<T> into(FixedWindows.of(DEFAULT_WINDOW_SIZE)));
}

代码示例来源：origin: org.apache.beam/beam-sdks-java-core

@Override
 public PCollection<String> expand(PCollection<String> in) {
  return in.apply(
      "Window",
      Window.<String>into(windowFn).withTimestampCombiner(TimestampCombiner.EARLIEST))
    .apply(Count.perElement())
    .apply("FormatCounts", ParDo.of(new FormatCountsDoFn()))
    .setCoder(StringUtf8Coder.of());
 }
}

内容来源于网络，如有侵权，请联系作者删除！

window

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com