混合对象和行，携带多个对象。如何从行数据集到对象数据集（数据集，其中 + 表示：“连接”）？

Question

下面的问题比较长。它描述了这一点：

• 我有一个由原始类型 a1、a2...a10、b1、b2...b8、c1..c4、d1 组成的行数据集。
• 它们隐藏对象 A、B、C，有时还伴有不在类中的其他基元属性：d1、例如。
• 我可以返回一个 Dataset 而不是 Dataset，其中 E 是一个具有 A 成员的类、B、C 对象和属性 d1。
• 但如果可以的话，我想避免这种情况，看看我可以在多大程度上找到一个解决方案，在该解决方案中我将返回连接对象[和属性]的数据集：Dataset< /代码>
（其中 + 符号表示对象通过连接链接在一起）。
如果可能的话，真的可以管理吗？

---

我的代码主要使用Dataset。

例如，我有一个构建（法语）城市描述的方法：

/**
* Obtenir un Dataset des communes.
* @param session Session Spark.
* @param anneeCOG Année du Code Officiel Géographique de référence.
* @param verifications Vérifications demandées.
* @return Dataset des communes.
* @throws TechniqueException si un incident survient.
*/
public Dataset<Row> rowCommunes(SparkSession session, int anneeCOG, Verification... verifications) throws TechniqueException {
  String nomStore = "communes_par_codeCommune";
  
  Dataset<Row> communes = loadFromStore(session, "{0}_{1,number,#0}", nomStore, anneeCOG, verifications);
  
  if (communes != null) {
     return communes;
  }

  LOGGER.info("Constitution du dataset des communes depuis pour le Code Officiel Géographique (COG) de l'année {}...", anneeCOG);
  
  Dataset<Row> c = loadAndRenameCommmunesCSV(session, anneeCOG, false, verifications);
  Dataset<Row> s = this.datasetSirenCommunaux.rowSirenCommunes(session, anneeCOG, TriSirenCommunaux.CODE_COMMUNE);

  Column condition1 = c.col("codeCommune").equalTo(s.col("codeCommune"));
  Column condition2 = c.col("codeCommuneParente").equalTo(s.col("codeCommune"));
  
  verifications("jonction communes et siren par codeCommune", c, null, s, condition1, verifications, SHOW_REJETS, COMPTAGES_ET_STATISTIQUES);
  Dataset<Row> join1 = c.join(s, condition1)
     .drop(s.col("codeCommune"))
     .drop(s.col("nomCommune"))
     .drop(s.col("codeRegion"))
     .drop(s.col("codeDepartement"));

  verifications("jonction communes et siren par codeCommune, join1", c, null, null, null, verifications);
  
  verifications("jonction communes et siren par codeCommuneParente", c, null, s, condition2, verifications, SHOW_REJETS, COMPTAGES_ET_STATISTIQUES);
  Dataset<Row> join2 = c.join(s, condition2)
     .drop(s.col("codeCommune"))
     .drop(s.col("nomCommune"))
     .drop(s.col("codeRegion"))
     .drop(s.col("codeDepartement"));

  verifications("jonction communes et siren par codeCommuneParente, join2", c, null, null, null, verifications);
  
  communes = join1.union(join2);

  // La strate communale doit concorder avec celle des comptes individuels des communes.
  communes = communes.withColumn("strateCommune", 
     when(s.col("populationTotale").between(0, 249), lit(1))       // communes de moins de 250 hab
     .when(s.col("populationTotale").between(250, 499), lit(2))     // communes de 250 à 500 hab
     .when(s.col("populationTotale").between(500, 1999), lit(3))    // communes de 500 à 2 000 hab
     .when(s.col("populationTotale").between(2000, 3499), lit(4))   // communes de 2 000 à 3 500 hab
     .when(s.col("populationTotale").between(3500, 4999), lit(5))   // communes de 3 500 à 5 000 hab
     .when(s.col("populationTotale").between(5000, 9999), lit(6))   // communes de 5 000 à 10 000 hab
     .when(s.col("populationTotale").between(10000, 19999), lit(7)) // communes de 10 000 à 20 000 hab
     .when(s.col("populationTotale").between(20000, 49999), lit(8)) // communes de 20 000 à 50 000 hab
     .when(s.col("populationTotale").between(50000, 99999), lit(9)) // communes de 50 000 à 100 000 hab
     .otherwise(lit(10)));                                          // communes de plus de 100 000 hab

  // Obtenir les contours des communes.
  // "(requête SQL) contours" est la forme de substitution pour Spark. cf https://stackoverflow.com/questions/38376307/create-spark-dataframe-from-sql-query
  String format = "(select insee as codecommuneosm, nom as nomcommuneosm, surf_ha as surface2, st_x(st_centroid(geom)) as longitude, st_y(st_centroid(geom)) as latitude from communes_{0,number,#0}) contours";
  String sql = MessageFormat.format(format, anneeCOG);
  
  Dataset<Row> contours = sql(session, sql).load();
  contours = contours.withColumn("surface", col("surface2").cast(DoubleType)).drop(col("surface2"))
     .orderBy("codecommuneosm");
  
  Column conditionJoinContours = col("codeCommune").equalTo(col("codecommuneosm"));
  
  verifications("jonction communes et contours communaux OSM (centroïde, surface)", communes, null, contours, conditionJoinContours, verifications, SHOW_REJETS, COMPTAGES_ET_STATISTIQUES);
  communes = communes.join(contours, conditionJoinContours, "left_outer")
     .drop(col("codecommuneosm")).drop(col("nomcommuneosm"));
  verifications("jonction communes et contours communaux OSM (centroïde, surface)", communes, null, null, null, verifications);

  // Associer à chaque commune son code intercommunalité, si elle en a un (les communes-communautés peuvent ne pas en avoir).
  Dataset<Row> perimetres = this.datasetPerimetres.rowPerimetres(session, anneeCOG, EPCIPerimetreDataset.TriPerimetresEPCI.CODE_COMMUNE_MEMBRE).selectExpr("sirenCommuneMembre", "sirenGroupement as codeEPCI", "nomGroupement as nomEPCI");
  Column conditionJoinPerimetres = communes.col("sirenCommune").equalTo(perimetres.col("sirenCommuneMembre"));

  verifications("jonction communes et périmètres", communes, null, perimetres, conditionJoinPerimetres, verifications, SHOW_REJETS, COMPTAGES_ET_STATISTIQUES);
  communes = communes.join(perimetres, conditionJoinPerimetres, "left");

  // Y associer les départements.
  communes = this.datasetDepartements.withDepartement(session, "codeDepartementRetabli", communes, "codeDepartement", null, true, anneeCOG)
     .drop("codeRegionDepartement")
     .drop("codeDepartementRetabli");

  communes = communes.repartition(col("codeDepartement"))
     .sortWithinPartitions(col("codeCommune"))
     .persist(); // Important : améliore les performances.

  saveToStore(communes, new String[] {"codeDepartement"}, "{0}_{1,number,#0}", nomStore, anneeCOG);
  LOGGER.info("Le dataset des communes du Code Officiel Géographique de l'année {} est prêt et stocké.", anneeCOG);

  return communes;
}

有时，如果我将这些行转换为 Commune 对象，会很有用，因为业务对象（至少在服务器端）可以具有方法这给他们带来了某种智慧（仅限于观察自己或包裹中的物品）。

例如，Commune 对象具有此方法，可以在可以在其名称中找到一篇文章时帮助检测它是否与另一个对象具有相同的名称。

/**
* Déterminer si notre commune a le même nom que celle en paramètre.
* @param nomCandidat Nom de commune : il peut contenir une charnière.
* @return true si c'est le cas.
*/
public boolean hasMemeNom(String nomCandidat) {
  // Si le nom soumis vaut null, répondre non.
  if (nomCandidat == null) {
     return false;
  }
  
  // Faire une comparaison directe de nom de commune tout d'abord, car l'emploi du collator est très coûteux.
  if (nomCandidat.equalsIgnoreCase(this.nomCommune)) {
     return true;
  }
  
  // Puis, rechercher avec les différentes charnières.
  if (nomCandidat.equalsIgnoreCase(nomAvecType(false, PrefixageNomCommune.AUCUN))) {
     return true;
  }
  
  if (nomCandidat.equalsIgnoreCase(nomAvecType(false, PrefixageNomCommune.A))) {
     return true;
  }
  
  if (nomCandidat.equalsIgnoreCase(nomAvecType(false, PrefixageNomCommune.POUR))) {
     return true;
  }
  
  // En cas d'échec, reprendre ces tests, mais avec le collator, plus lent, mais qui passera outre les caractères accentués.
  if (collator.equals(nomCandidat, this.nomCommune)) {
     return true;
  }
  
  if (collator.equals(nomCandidat, nomAvecType(false, PrefixageNomCommune.AUCUN))) {
     return true;
  }
  
  if (collator.equals(nomCandidat, nomAvecType(false, PrefixageNomCommune.A))) {
     return true;
  }
  
  if (collator.equals(nomCandidat, nomAvecType(false, PrefixageNomCommune.POUR))) {
     return true;
  }
  
  return false;
}

为了实现从 Dataset 到 Dataset 的转换，我编写了这个函数，它可以工作，但让我很麻烦，因为它对我来说看起来很笨拙：

/**
* Obtenir un Dataset des communes (comme objets communes, incluant les données siren communaux).
* @param session Session Spark.
* @param anneeCOG Année du Code Officiel Géographique de référence.
* @return Dataset des communes.
* @throws TechniqueException si un incident survient.
*/
public Dataset<Commune> obtenirCommunes(SparkSession session, int anneeCOG) throws TechniqueException {
  Dataset<Row> communes = rowCommunes(session, anneeCOG);
  
  return communes
     .select(communes.col("typeCommune"), communes.col("codeCommune"), communes.col("codeRegion"), communes.col("codeDepartement"),
         communes.col("arrondissement"), communes.col("typeNomEtCharniere"), communes.col("nomMajuscules"), communes.col("nomCommune"), 
         communes.col("codeCanton"), communes.col("codeCommuneParente"), communes.col("sirenCommune"), communes.col("populationTotale").alias("population"),
         communes.col("strateCommune"), communes.col("codeEPCI"), communes.col("surface"), communes.col("longitude"), communes.col("latitude"),
         communes.col("nomDepartement"), communes.col("nomEPCI"))
     .as(Encoders.bean(Commune.class))
     .cache();
}

但是，我的主要问题是：

此后，在大多数数据集中，Commune 对象（或行）几乎从未单独使用（或返回）。

通常，Commune 会附带一些相关的就业数据或相关的财务信息。或者有时，如果有意义的话，计算方法可以仅将一两个原始（原始）数据值与城市相关联。

那么，直到今天，到底发生了什么？

• 我从一个包含 20 列的 Dataset 开始。

• 如果我正在进行一些连接、计算、丰富等操作：该数据集有时会达到 40 或 50 列，

  • 假设一条记录是：具有原始类型的行列的 a1, a2...a10, b1, b2...b8, c1..c4, d1。

• 然后，我通过 Encoders.bean(...) 方法从该记录中提取它隐藏的各种业务对象，具体取决于我的需要。

  • 从记录中，我可以从 a1、a2...a10 中提取 A 或从 c1... 中提取 C c4.

它起作用了（而且正在起作用），但我并不为这种做事方式感到自豪。

我更喜欢从一个简单的 Dataset 数据集开始，向我的 API 用户完全隐藏行阶段，然后能够根据他们的需求返回给他们：

• a数据集同时包含 Commune 和 Employment 业务对象，或 Commune 和 Accounting。

• 包含 Commune 的数据集，但也包含 "d1"、"h1" 和 "k1 列的少量行值“（如果需要进行一些计算来提供该城市的特定信息，出于某些特殊目的/情况，不会导致更改整个业务对象描述，但仅在此时返回额外的列值除了城市描述之外）。

这意味着我会遇到我想返回显示每个“记录”的数据集的情况：
A, B（具体对象聚集在一起）
或
A、C
甚至有时A,B,C。

或者说我最担心的是：
A、C、d1（两个具体对象，加上...一个原始值）。

在我开始重大行动之前，您能否给我一些关于如何处理此类问题的想法？

---

警告：这个问题不是一个简单的问题。它甚至可能没有一个明确的解决方案。

• 从由基本类型a1, a2...a10, b1, b2...b8, c1..c4, d1组成的行记录开始，我能够提取A、B、C 对象（如果我同意对 nn 进行最多 n 次额外转换） em> 我可以在我的中找到不同的对象类型属性。

• 但我想知道如何尽可能接近这种情况，考虑到 + 符号意味着：加入，我有一个数据集：
  数据集
  甚至：Dataset，其中 d1 仍然是不在对象中的原始类型。

• 如果实现了这一点，可能会带来麻烦。
  什么是数据集.foreach？
  A、B 或 C 对象？

之后你会如何处理？

我还生活在属性的熔岩中，我想知道是否可以使用对象将我的数据集改进为新的数据集。绑定在一起以描述单个记录的对象。

---

这个问题可以通过创建一个 E 对象来解决，该对象具有成员 A、B 或 C 对象以及d1 原始属性，并返回 Dataset。

但此时我想尽我所能避免这种情况。首先尝试找出我还能做些什么。