javascript - Node.js で準備ができたらデータをパイプする

Question

次のような大きな JSON ファイルがあります。

[
 {
  "id": 2000,
  "city": "New York",
  "lat": "",
  "lon": "",
 },
...
]

そして、各オブジェクトの緯度と経度を探しています。データをパイプするために JSONStream モジュールを使用しています。

var writeStream = fs.createWriteStream('data/stream.json');
var readStream = fs.createReadStream('data/dummy/institucion_1_2000.json', {encoding: 'utf8'})

// Search for lat and lon
var search = es.mapSync(function(data){
  if(data.lat == ""){
    // Search for the lat and lon
    geocoder.geocode({address: data.city, country: "US"}) 
      .then(function(res) {
        console.log("Searched for " + res[0].formattedAddress);
        data.lat = res[0].latitude;
        data.lon = res[0].longitude;
      })
      .catch(function(err) {
        console.log("There was an error with element with id = " + data.id);
        console.log("Here is the error: " + err);
        if(err == 'Error: Status is OVER_QUERY_LIMIT. You have exceeded your rate-limit for this API.') {
          process.exit();
        }
      });
    return data;
  }
})

// Pipe
readStream
  .pipe(JSONStream.parse('*'))
  .pipe(search) 
  .pipe(JSONStream.stringify()) // This doesent wait until the search is finish
  .pipe(writeStream)

ジオコーディング部分が機能します。

私の問題は、検索関数が終了する前に JSONStream.stringify がデータを読み取ってパイプ処理することです。したがって、必要な変更を加えずに同じ JSON ファイルを取得しています。これを試してみると：

if(data.lat == ""){
    lat = 1;
}

より多くの時間がかかるジオコーディングの代わりに、それは機能します。私の問題は、ストリーミングされたデータを変更するのにかかる時間にあると思います。では、変更されたデータをパイプする方法はありますか?

編集 同期と非同期の間で混乱がありました。ジョンズのおかげで

var search = es.map(function (data, callback) {
  if(data.lat == ""){
    geocoder.geocode({address: data.city, country: "USA"}) // Choose between Comuna_Empresa and Comuna_Institucion
    .then(function(res) {
      console.log("Searched for " + res[0].formattedAddress);
      data.lat = res[0].latitude;
      data.lon = res[0].longitude;
      callback(null,data);
    })
    .catch(function(err) {
      console.log("There was an error with element at index = " + index);
      console.log("Here is the error: " + err);
    });
  }
})