实现流云数据流转换，为流中的每个元素调用API

Question

我相信，这不是实施这种转变的最佳方式。我想知道是否有更好的方法从数据流作业中进行查找。

跟进：

试图将用例实现为侧输入：

> PcollectionView<DatabaseReader> reader = pipeline.apply(
>        ParDo.of(
>                new DoFn<Long, DatabaseReader>() {
>                    @ProcessElement
>                    public void process(
>                            @Element Long input, 
    OutputReceiver<DatabaseReader> o) throws IOException { //Following
> MaxMind Documentation to read file
>                            File database = new File(“GeoIP2-City.mmdb”);
>                            DatabaseReader reader = new DatabaseReader
>                                    .Builder(database)
>                                    .fileMode(com.maxmind.db.Reader.FileMode.MEMORY)
>                                    .withCache(new CHMCache())
>                                    .build();
>                            o.output(reader);
>                    }
>                })) .apply(View.asSingleton());

> PCollection<TableRow> newTableRow =   
> tableRowBeforeTransformation.apply("mmdb lookup", ParDo.of(new
> MmdbReaderFn(reader)).withSideInputs(reader));

> DatabaseReader reader; PCollectionView<DatabaseReader> readerView;
> 
> public MmdbReaderFn(PCollectionView readerView){    this.readerView =
> readerView;
> 
> }
> 
> @ProcessElement public void processElement(ProcessContext context)
> throws IOException {    TableRow tableRow = context.element();   
> reader = context.sideInput(readerView);    String ip =
> String.valueOf(tableRow.get("ip"));    try {
>        InetAddress ipAddress = InetAddress.getByName(ip);
> 
>        CityResponse response = reader.city(ipAddress);
>        if(response == null) {
>            log.info("Invalid IP address");
>        } else {
> 
>            if(response.getCountry() != null && response.getCountry().getName() != null) {
>                tableRow.set(COUNTRY_NAME, response.getCountry().getName());      // 'Unite States'
>            } else {
>                tableRow.set(COUNTRY_NAME, null);
>            }
> 
>           
>            if(response.getCity() != null && response.getCity().getName() != null) {
>                tableRow.set(CITY_NAME, response.getCity().getName());
>            } else {
>                tableRow.set(CITY_NAME, null);
>            }
> 
>                } catch (Exception e) {
>        log.error("Error process GeoIP2", e);
> 
>    }    context.output(tableRow); }

Answer 1

如果预计130MB文件将来不会增长，您可以使用SideInput，这将避免管道外的rpc调用。

请注意，侧输入确实需要适合工人的内存，本文提供了一些关于内存需求分析的好内容。

如果文件定期更改，则可以使用缓慢更新的侧输入模式。