如何将3个流中的数据聚合到一个组合对象中？

Question

我在一个Apache Flink项目中遇到了以下情况。

具有不同对象的3个流，例如

Person->字符串id,字符串firstName,字符串lastName（即101,John,Doe）

PersonDetail->字符串id,字符串地址,字符串城市,字符串电话号码,long personId（即99,Stefansplatz 1,+43066012345678,101）

PersonAddDetail->字符串id、字符串AddDetailType、对象AddDetailValue、长personId（即77,1,Hansi或78,2,1234或80,3,true）

我想将这些流中的对象聚合（不确定这里的措词是否正确）到一个新对象，然后放入一个新流中。聚合应该基于Person id，作为附加的catch，我需要仅用特定的AddDetailType筛选出PersonAddDetailType（假设我只对类型为1和2的对象感兴趣）。

聚合对象看起来应该像

PersonReport->长id、字符串名字、字符串名字、字符串地址、字符串城市、字符串电话号码、ArrayList 详细信息

现在的问题是，这是否可能，如果是，我该如何完成。欢迎输入。

Answer 1

感谢@Jeremy Grand comment，我自己想出了一个解决方案，我想分享我的想法和代码。我引入了一个名为PersonContainer的新类

public class PersonContainer {

private String id;

private Person person;
private PersonDetail personDetail;
private List<PersonAddDetail> personAddDetailList = new ArrayList<>();

public PersonContainer(Person person) {
  this.id = person.getID();
  this.person = person;
}



public PersonContainer(PersonDetail personDetail) {
    this.id = personDetail.getOTTRID();
    this.personDetail = personDetail;
  }

  public PersonContainer(PersonAddDetail personAddDetail) {
    this.id = personAddDetail.getOTTRID();
    this.timeStamp = ttrDetailAddEvent.getDATECREATED();
    this.personAddDetailList.add(personAddDetail);
  }

  public PersonContainer merge(PersonContainer other) {
    if (other.person != null) {
      this.person = other.person;
      return this;
    }
    if (other.personDetail != null) {
      this.personDetail = other.personDetail;
      return this;
    }
    if (other.personAddDetailList.size() > 0) {
      this.personAddDetailList.addAll(other.personAddDetailList);
      return this;
    }
    return null;
  }

  public String getId() {
    return id;
  }

  public Person getPerson() {
    return person;
  }

  public PersonDetail getPersonDetail() {
    return personDetail;
  }

  public List<PersonAddDetail> getPersonAddDetailList() {
    return PersonAddDetailList;
  }

  public boolean isComplete() {
    return person != null && personDetail != null && personAddDetailList.size() > 1;
  }
}

这是很重要的部分，因为我首先要将三个输入流的对象映射到这个公共对象，然后再联合流。

下面是我所做的，我在评论中描述了单个步骤。简而言之，我将三个输入流映射到新引入容器的新流。然后，我对这三个流进行联合，并使用迭代模式对这些对象进行键控，并使用我的自定义merge方法合并它们。最后，我定义了一个自定义的complete方法来区分完全合并的容器，这些容器最终映射到输出，而未完成的容器则反馈到合并过程中。

//Filter PersonAddDetail to have just the types needed
DataStream<PersonContainer> filteredPersonAddDetail = unfilteredPersonAddDetail.filter(new FilterFunction<OboTtrDetailAddEvent>() {
      @Override
      public boolean filter(PersonAddDetail personAddDetail) throws Exception {
        return personAddDetail.getAddDetailType().matches("1|2");
      }
    });

//map Person stream to common object
DataStream<PersonContainer> mappedPersonStream = personInputStream.map(new MapFunction<Person, PersonContainer>() {
  @Override
  public PersonContainer map(Person Person) throws Exception {
    return new PersonContainer(Person);
  }
});

//map PersonDetail stream to common object
DataStream<PersonContainer> mappedPersonDetailStream = personDetailInputStream.map(new MapFunction<PersonDetail, PersonContainer>() {
  @Override
  public PersonContainer map(PersonDetail PersonDetail) throws Exception {
    return new PersonContainer(PersonDetail);
  }
});

//map PersonAddDetail stream to common object
DataStream<PersonContainer> mappedPersonAddDetailStream = filteredPersonAddDetail.map(new MapFunction<PersonAddDetail, PersonContainer>() {
  @Override
  public PersonContainer map(PersonAddDetail PersonAddDetail) throws Exception {
    return new PersonContainer(PersonAddDetail);
  }
});

//union the three input streams to one single stream
DataStream<PersonContainer> combinedInput = mappedPersonStream.union(mappedPersonDetailStream, mappedPersonAddDetailStream);

// Iteration pattern is in place here and I'm going to recursively try to merge corresponding objects togehter
IterativeStream<PersonContainer> iteration = combinedInput.iterate();

// Group objects by there shared ID and then use reduce to merge them
DataStream<PersonContainer> iterationBody = iteration.keyBy(new KeySelector<PersonContainer, String>() {
  @Override
  public String getKey(PersonContainer personContainer) throws Exception {
    return personContainer.getId();
  }
})
    .reduce(new ReduceFunction<PersonContainer>() {
      @Override
      public PersonContainer reduce(PersonContainer personContainer, PersonContainer other) throws Exception {
        return personContainer.merge(other);
      }
    });

// use the containers complete method to check whether the merge is finished or we need to wait for further objects in the stream   
DataStream<PersonContainer> containersNotCompleteYet = iterationBody.filter(new FilterFunction<PersonContainer>() {
  @Override
  public boolean filter(PersonContainer PersonContainer) throws Exception {
    return !personContainer.isComplete();
  }
});

// partially merged or not merged at all containers are put back on the stream
iteration.closeWith(containersNotCompleteYet);

// fully merged containers are processed further
DataStream<PersonContainer> completeContainers = iterationBody.filter(new FilterFunction<PersonContainer>() {
  @Override
  public boolean filter(PersonContainer PersonContainer) throws Exception {
    return personContainer.isComplete();
  }
});

// finally the container is mapped to the correct output object
DataStream<PersonReport> personReport = completeContainers.map(new MapFunction<PersonContainer, PersonReport>() {
  @Override
  public PersonReport map(PersonContainer personContainer) throws Exception {

    // map personContainer to final PersonReport

    return personContainer;
  }
});

这种方法对我很有效，好的是我可以处理在流中晚到达的对象（比方说PersonAddDetail在其他对象之后几分钟到达），而且我不需要定义某种窗口。谢谢你的投入

Answer 2

您的问题听起来像是join操作。您可以执行以下操作：

personDataStream.join(personDetailDataStream).where(new KeySelector<Person, Long>() {
    ...
}).equalTo(new KeySelector<PersonDetail, Long>() {
    ...
}).window(TumblingEventTimeWindows.of(Time.seconds(2))).apply(new JoinFunction<Person, PersonDetail, PersonWithDetail>() {
   ...
});

注意，通常情况下，连接操作是不可能在无界（无限）集合上进行的，所以您需要将其绑定到Windows中。