java - 从 google dataflow 输出到 google cloud firestore

Question

美好的一天，经过大量努力尝试掌握这项技术(数据流)后，我设法使管道 100% 正常工作。

它所做的是将一堆 CSV 文件加载到管道中(来自谷歌云存储)，将它们转换为“域”对象，然后将它们以 JSON 格式保存到文件中。

不过，我想做的是获取 JSON 对象并将其直接推送到数据库 (google cloud firestore)。

我在这个阶段应用到我的数据的最终转换是:

.apply(DatastoreIO.v1().write().withProjectId("____"));

据我了解，调用需要先前的转换才能返回一个我无法创建的实体对象

public Entity toEntity() {
    Datastore datastore = DatastoreOptions.getDefaultInstance().getService();
    Key taskKey = datastore.newKeyFactory().setKind("Task").newKey("Test");
    Entity e = Entity.newBuilder(taskKey).set("Domain", domain)
            .set("LocationOnsite", locOnSite)
            .set("Company", company).build();

    return e;
}

这会返回一个 com.google.cloud.datastore.Entity 而不是所需的 com.google.datastore.v1.Entity

我认为值得注意的是，“Domain”对象还包含一些其他对象的 ArrayList，例如需要包含到数据库中的“Emails”。

下面是我目前拥有的示例 JSON 输出:

{
   "Vertical": "Business And Industrial",
   "Zip": "35229",
   "Company": "Alabama Association of Nonprofits",
   "QuantCast": "229219",
   "Twitter": "",
   "Vimeo": "",
   "LocationOnSite": "",
   "LastIndexed": "2018-02-01",
   "Pinterest": "",
   "Youtube": "",
   "TechSpend": "$250+",
   "Emails": [
      {
         "Email": "shannon@alabamanonprofits.org"
      },
      {
         "Email": "support@alabamanonprofits.org"
      },
      {
         "Email": "carla@alabamanonprofits.org"
      },
      {
         "Email": "kellie@alabamanonprofits.org"
      },
      {
         "Email": "ashley@alabamanonprofits.org"
      },
      {
         "Email": "Unknown"
      }
   ],
   "Facebook": "",
   "Google+": "",
   "Alexa": "",
   "Github": "",
   "FirstIndexed": "2011-01-03",
   "People": [
      {
         "Email": "Unknown",
         "Name": "Joshua Cirulnick"
      },
      {
         "Email": "Unknown",
         "Position": "Other",
         "Name": " Elaine Lin"
      },
      {
         "Email": "Unknown",
         "Position": "Other",
         "Name": " Terry Burkle"
      },
      {
         "Email": "Unknown",
         "Position": "Director",
         "Name": " Ashley Gilbert"
      },
      {
         "Email": "Unknown",
         "Position": "President",
         "Name": " Carol Weisman"
      },
      {
         "Email": "Unknown",
         "Position": "Csuite",
         "Name": " Shannon Ammons"
      },
      {
         "Email": "Unknown",
         "Position": "Founder",
         "Name": " Kelly McDonald"
      }
   ],
   "City": "Birmingham",
   "Telephone#s": [
      {
         "Telephone#": "+1-205-879-4712"
      },
      {
         "Telephone#": "+1-205-871-7740"
      }
   ],
   "FirstDetected": "N/A",
   "LinkedIn": "",
   "VK": "",
   "State": "AL",
   "Instagram": "",
   "Country": "US",
   "Domain": "alabamanonprofits.org",
   "LastFound": "N/A"
}

如果有人能为我指出正确的方向，让我知道如何有效地将这些对象放入谷歌云 firestore 数据库中，我会非常高兴!

Answer 1

您可以将数据写入 Cloud Pub/Sub，这会触发一个函数将数据写入 Cloud Firestore。在 Google I/O 2017 上有一个很好的例子，它做同样的事情，但使用实时数据库。

您可以在这里观看:Data Pipelines with Firebase and Google Cloud (Google I/O '17)