0 
Pattern< Tuple3< String, String, String >, ? > pattern = Pattern.<Tuple3< String, String, String > > begin( "start" )            
            .next( "3" ).where( new FilterFunction< Tuple3< String, String, String > >() {
                @Override
                public boolean filter ( Tuple3< String, String, String > value ) throws Exception {
                    return value.f2.equals( "3" );
                }
            } )
            .next( "4" ).subtype(Tuple.getTupleClass( 2 )).where( new FilterFunction< Tuple2< String, String> >() {
                @Override
                public boolean filter ( Tuple2< String, String > value ) throws Exception {
                    return value.f1.equals( "3" );
                }
            } )

subtype(Tuple.getTupleClass( 2 ))，并出现错误 Inferred type 'capture<? extends org.apapche.flink.api.java.tuple.Tuple>' for type parameter 'S' is not within its bound;should extend 'org.apapche.flink.api.java.tuple.Tuple3<java.lang.String,java.lang.String,java.lang.String>'

我应该修改这个吗？但是如何修改？Pattern< Tuple3< String, String, String >, ? > pattern

由 2017012 更新

JoinedStreams< Tuple2< String, String >, Tuple3< String, String, String > >.Where< String >.EqualTo
        joinedStreams = someStream
        .join( otherStream )
        .where( value -> value.f1 )
        .equalTo( value -> value.f1 );

Pattern< Tuple, ? > pattern = Pattern.< Tuple > begin( "start" )
        .subtype( Tuple3.class )
        .where( evt -> evt.f2.equals( "3" ) )
        .next( "4" )
        .subtype( Tuple2.class )
        .where( evt -> evt.f1.equals( "3" ) )
        .within( Time.seconds( 10 ) );

PatternStream< ...> patternStream = CEP.pattern( joinedStreams, pattern );

我试过这个，不应该填写什么。PatternStream< ...>感谢任何可以提供帮助的人。

4

2 回答 2

1

试试这个代码：

Pattern<Tuple, ?> pattern =
    Pattern.<Tuple>begin("start")
    .next("3")
        .subtype(Tuple3.class)
        .where(new FilterFunction<Tuple3>() {

            @Override
            public boolean filter(Tuple3 value) throws Exception {
                return value.f2.equals("3");
            }
        })
    .next("4")
        .subtype(Tuple2.class)
        .where(new FilterFunction<Tuple2>() {

            @Override
            public boolean filter(Tuple2 value) throws Exception {
                return value.f1.equals("3");
            }
        });

从通用类型开始Tuple，使用具体类型Tuple2和Tuple3子事件。这种模式的数据流必须有一个Tuple类型。

于 2017-01-10T11:48:04.600 回答
0

那这个呢：

    Pattern<Tuple, ?> pattern = Pattern.<Tuple>begin("start")
            .subtype(Tuple3.class)
            .where(evt -> evt.f2.equals("3"))
            .next("4")
            .subtype(Tuple2.class)
            .where(evt -> evt.f1.equals("3"))
            .within(Time.seconds(10));
  1. 你不需要在开始后添加下一个
  2. 注意子类型的字面意思，tuple3 和 tuple2 应该扩展 tuple。

如果要连接两个不同的数据流。

DataStream<Tuple2> someStream = //...
DataStream<Tuple3> otherStream = //...

ConnectedStreams<Tuple2, Tuple3> connectedStreams = someStream.connect(otherStream);

然后可以使用 CoMap、CoFlatMap 得到相同的类型，例如将 Tuple2、Tuple3 转换为 String：ConnectedStreams → DataStream

connectedStreams.flatMap(new CoFlatMapFunction<Tuple2, Tuple3, String>() {

   @Override
   public void flatMap1(Integer value, Collector<String> out) {
       out.collect(Tuple2.toString());
   }

   @Override
   public void flatMap2(String value, Collector<String> out) {
       for (String word: value.split(" ")) {
         out.collect(Tuple3.toString);
       }
   }
});

以下是一些有用的链接，介绍了一个很好的用例：

  1. 使用 Apache Flink 介绍复杂事件处理 (CEP)
  2. 我翻译的中文版
于 2017-01-10T03:21:05.793 回答