小编Anu*_*nup的帖子

我正在使用 Windows 7 机器来测试 NodeJS 中的集群。

代码如下：

var cluster = require('cluster');
var http = require('http');
var numCPUs = require('os').cpus().length;

     if (cluster.isMaster) { 
     // Fork workers.
     for (var i = 0; i < numCPUs; i++) {
        cluster.fork();
     }

     cluster.on('exit', function (worker, code, signal) {
        console.log('worker ' + worker.process.pid + ' died');
     });

     cluster.on('listening', function(worker, address) {
        console.log("A worker with #" + worker.id + " is now connected to " + address.address + ":" + address.port);        
     });
   } else {
      // Workers …

我试图使用Apache Beam创建一个Dataflow管道,我无法按照文档,找不到任何示例.

管道很简单.

1. 创建一个管道
2. 阅读pub/sub主题
3. 写入扳手.

目前,我陷入了第2步.我找不到任何关于如何从pub/sub读取并使用它的示例.

这是我到目前为止所希望的代码

class ExtractFlowInfoFn extends DoFn<PubsubMessage, KV<String, String>> {
    public void processElement(ProcessContext c) {
        KV.of("key", "value");
    }
}

public static void main(String[] args) {

    Pipeline p = Pipeline.create(
    PipelineOptionsFactory.fromArgs(args).withValidation().create());

    p.apply("ReadFromPubSub", PubsubIO.readMessages().fromSubscription("test"))
     .apply("ConvertToKeyValuePair", ParDo.of(new ExtractFlowInfoFn()))
     .apply("WriteToLog", ));
};

我通过以下几个例子来提出代码.说实话,我不知道我在这做什么.

请帮助我理解这一点或将我链接到正确的文档.