java - groupByKey(...) 中的 类在其成员中有一个 Map。 groupByKey 操作因 "un-comparable"问题而失败

Question

我有课Entreprise具有基元数据类型和另一个类上的映射:Etablissement它仅由原始数据类型组成。

public class Entreprise implements Comparable<Entreprise> {
   /** Liste des établissements de l'entreprise. */
   private Map<String, Etablissement> etablissements = new HashMap<>();

   /** Sigle de l'entreprise */
   private String sigle;

   /** Nom de naissance */
   private String nomNaissance;

   /** Nom d'usage */
   private String nomUsage;
 ...
   @Override
   public int compareTo(Entreprise o) {
      return getSiren().compareTo(o.getSiren());
   }
}

在EntrepriseDataset中我创建了这个架构:

public StructType schemaEntreprise() {
   StructType schema = new StructType()
      .add("sigle", StringType, true)
      .add("nomNaissance", StringType, true)
      .add("nomUsage", StringType, true)
       ...

   // Ajouter au Dataset des entreprises la liaison avec les établissements.
   MapType mapEtablissements = new MapType(StringType, this.datasetEtablissement.schemaEtablissement(), true);
   StructField etablissements = new StructField("etablissements", mapEtablissements, true, Metadata.empty());
   schema.add(etablissements);

  return schema;
}

我可以做便利joinWith例如:

Dataset<Tuple2<Entreprise, Etablissement>> ds = dsEntreprises
    .joinWith(dsEtablissements,
    dsEntreprises.col("siren").equalTo(dsEtablissements.col("siren")), "inner");

这是一些操作的开始，应该引导我到 Entreprise 的数据集对象有其 Etablissement map 中的对象

Enterprise : {{834935512, Activité principale : 68.20A (NAFRev2), effectif salarié : null (null, employeur : null), active : null, dernier traitement : Jan 26, 2018, historisation débutée le Dec 26, 2017, nombre de périodes sans changement : 1}, nombre d'établissements : 1, catégorie entreprise : null (null), catégorie juridique : 1000, n° répertoire national des associations : null, Economie Sociale et Solidaire : null, NIC de l'établissement siège : 00014, sigle : null, dénomination de l'entreprise : {18}, dénominations usuelles 1 : null, 2 :{19}, 3 : {20}, 4 : {21} , Nom de naissance : LOHIER, Nom d'usage : null, prénom usuel : ROGER, autres prénoms : ROGER, pseudonyme : null, sexe : M, purgée : null, date de création : Dec 26, 2017}
Etablishment : {{83493551200014, Activité principale : 68.20A (NAFRev2), effectif salarié : null (null, employeur : null), active : null, dernier traitement : Jan 26, 2018, historisation débutée le Dec 26, 2017, nombre de périodes sans changement : 1}, activité au registre des métiers : null, date de création de l'établissement : 2017-12-26, établissement siège : false, dénomination de l'établissement : null, enseigne 1 : null, 2 : null, 3 : null, adresses : {anomalies : [], annulé logiquement : false, distribution spéciale : null, numéro dans la voie : 74, répétition : null, type de voie : BD, libellé de voie : DE LA PORTELETTE, complément d'adresse : 74-78, code postal : 80100, cedex : null - null, commune : 80001 - ABBEVILLE, commune étrangère : null, pays : null - null}}

Enterprise : {{178001111, Activité principale : 84.23Z (NAFRev2), effectif salarié : 41 (2016, employeur : null), active : null, dernier traitement : Jan 17, 2019, historisation débutée le Jan 1, 2008, nombre de périodes sans changement : 5}, nombre d'établissements : 3, catégorie entreprise : ETI (2,016), catégorie juridique : 7171, n° répertoire national des associations : null, Economie Sociale et Solidaire : null, NIC de l'établissement siège : 00016, sigle : null, dénomination de l'entreprise : {18}, dénominations usuelles 1 : null, 2 :{19}, 3 : {20}, 4 : {21} , Nom de naissance : null, Nom d'usage : null, prénom usuel : null, autres prénoms : null, pseudonyme : null, sexe : null, purgée : null, date de création : Jan 1, 1978}
Etablishment : {{17800111100396, Activité principale : 84.23Z (NAFRev2), effectif salarié : 11 (2016, employeur : null), active : null, dernier traitement : Sep 29, 2018, historisation débutée le Jan 1, 2008, nombre de périodes sans changement : 3}, activité au registre des métiers : null, date de création de l'établissement : 1983-01-01, établissement siège : false, dénomination de l'établissement : null, enseigne 1 : TRIBUNAL D'INSTANCE D'ABBEVILLE, 2 : null, 3 : null, adresses : {anomalies : [], annulé logiquement : false, distribution spéciale : BP A8, numéro dans la voie : 79, répétition : null, type de voie : RUE, libellé de voie : MARECHAL FOCH, complément d'adresse : null, code postal : 80100, cedex : 80103 - ABBEVILLE CEDEX, commune : 80001 - ABBEVILLE, commune étrangère : null, pays : null - null}}
Etablishment : {{17800111100743, Activité principale : 84.23Z (NAFRev2), effectif salarié : null (null, employeur : null), active : null, dernier traitement : Sep 1, 2008, historisation débutée le Dec 25, 2007, nombre de périodes sans changement : 1}, activité au registre des métiers : null, date de création de l'établissement : 2007-12-25, établissement siège : false, dénomination de l'établissement : null, enseigne 1 : TRIBUNAL PARITAIRE BAUX RURAUX, 2 : null, 3 : null, adresses : {anomalies : [], annulé logiquement : false, distribution spéciale : BP 330, numéro dans la voie : 79, répétition : null, type de voie : RUE, libellé de voie : MARECHAL FOCH, complément d'adresse : null, code postal : 80100, cedex : 80103 - ABBEVILLE CEDEX, commune : 80001 - ABBEVILLE, commune étrangère : null, pays : null - null}}
Etablishment : {{17800111100503, Activité principale : 84.23Z (NAFRev2), effectif salarié : 02 (2016, employeur : null), active : null, dernier traitement : Sep 29, 2018, historisation débutée le Jan 1, 2008, nombre de périodes sans changement : 3}, activité au registre des métiers : null, date de création de l'établissement : 1982-07-01, établissement siège : false, dénomination de l'établissement : null, enseigne 1 : CONSEIL DE PRUD'HOMMES D'ABBEVILLE, 2 : null, 3 : null, adresses : {anomalies : [], annulé logiquement : false, distribution spéciale : null, numéro dans la voie : 9, répétition : null, type de voie : AV, libellé de voie : DU GENERAL LECLERC, complément d'adresse : null, code postal : 80100, cedex : null - null, commune : 80001 - ABBEVILLE, commune étrangère : null, pays : null - null}}

但问题是我无法执行 groupByKey :

KeyValueGroupedDataset<Entreprise, Tuple2<Entreprise, Etablissement>> dsK = 
ds.groupByKey((MapFunction<Tuple2<Entreprise, Etablissement>, Entreprise>) f -> {
      Entreprise entreprise = f._1();
      Etablissement etablissement = f._2();
      entreprise.ajouterEtablissement(etablissement);

      return entreprise;
  }, 
  Encoders.bean(Entreprise.class));

Dataset<Entreprise> dsEntreprisesAvecEtablissements = 
dsK.mapGroups(new MapGroupsFunction<Entreprise, Tuple2<Entreprise, Etablissement>, Entreprise>() {
     @Override
     public Entreprise call(Entreprise key, Iterator<Tuple2<Entreprise, Etablissement>> values) {
        while(values.hasNext()) {
           Etablissement etablissement = values.next()._2();
           key.ajouterEtablissement(etablissement);
        }

        return key;
      }
   },
   Encoders.bean(Entreprise.class));

那个groupByKey(...) (或更准确地说，对 dsEntreprisesAvecEtablissements 数据集采取的第一个操作)失败并显示一条消息:

java.lang.IllegalArgumentException: cannot generate compare code for un-comparable type: map<string,struct<activiteArtisanRegistreDesMetiers:string,activitePrincipale:string,anneeValiditeEffectifSalarie:int,cedex:string,cedexSecondaire:string,codeCommune:string,codeCommuneSecondaire:string,codePaysEtranger:string,codePaysEtrangerSecondaire:string,codePostal:string,codePostalSecondaire:string,complementAdresse:string,complementAdresseSecondaire:string,dateCreationEtablissement:string,dateDebutHistorisation:string,dateDernierTraitement:string,denominationEtablissement:string,distributionSpeciale:string,distributionSpecialeSecondaire:string,enseigne1:string,enseigne2:string,enseigne3:string,indiceRepetition:string,indiceRepetitionSecondaire:string,libelleCedex:string,libelleCedexSecondaire:string,libelleVoie:string,libelleVoieSecondaire:string,nomCommune:string,nomCommuneSecondaire:string,nomPaysEtranger:string,nomPaysEtrangerSecondaire:string,nombrePeriodes:int,nomenclatureActivitePrincipale:string,numeroVoie:string,numeroVoieSecondaire:string,siege:boolean,siret:string,trancheEffectifSalarie:string,typeDeVoie:string,typeDeVoieSecondaire:string>>
    at org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext.genComp(CodeGenerator.scala:700) ~[spark-catalyst_2.12-2.4.3.jar:2.4.3]
    at org.apache.spark.sql.catalyst.expressions.codegen.GenerateOrdering$.$anonfun$genComparisons$3(GenerateOrdering.scala:121) ~[spark-catalyst_2.12-2.4.3.jar:2.4.3]

我发现没有其他 groupByKey 方法允许提供替代比较方法，即 groupByKey 的 equals(Map) 似乎在这里执行。它不认为我的 Java Pojo 目标为 Encoders.bean(..)是 Comparable 。所以，最终，我必须以这个笨拙的代码结束:

Entreprises entreprises = new Entreprises();

List<Tuple2<Entreprise, Etablissement>> tuples = ds.collectAsList();
Iterator<Tuple2<Entreprise, Etablissement>> itTuples = tuples.iterator();

while(itTuples.hasNext()) {
   Tuple2<Entreprise, Etablissement> tuple = itTuples.next();
   Entreprise entreprise = entreprises.get(tuple._1().getSiren());
   Etablissement etablissement = tuple._2();

   if (entreprise == null) {
      entreprise = tuple._1();
      entreprises.add(entreprise);
   }

   entreprise.ajouterEtablissement(etablissement);
}

return entreprises;

但是可能存在更好的方法来更好地解决我的问题。我应该做什么才能在操作结束时获得 Dataset<Entreprise>其中每个 Entreprise有他们的 map Etablissement在上面吗？

Answer 1

Spark 不允许比较 MapType。你可以做一些不同的事情。

关于您的代码，最重要的是用于加入和分组的键。这两个操作都是一样的。这使事情变得更加容易。

您可以尝试以下方法之一:

• 将 key 从 Enterprise 更改为 siren: String。并收集mapGroups中的所有Etablissement记录。如果 Enterprise 中出现重复 key ，这可能会导致问题。
• 在加入之前在 Etablissement 流中按 siren 进行分组，并将它们收集到 mapGroups 函数中。生成的流与Enterprise 流连接，后跟一个映射。

第一个解决方案

Dataset<Tuple2<Entreprise, Etablissement>> ds = dsEntreprises
                .joinWith(dsEtablissements,
                        dsEntreprises.col("siren").equalTo(dsEtablissements.col("siren")), "inner");

        KeyValueGroupedDataset<String, Tuple2<Entreprise, Etablissement>> dsK = ds.groupByKey((MapFunction<Tuple2<Entreprise, Etablissement>, String>)
                value -> value._1.siren, Encoders.STRING());

        dsK.mapGroups((MapGroupsFunction<String, Tuple2<Entreprise, Etablissement>, Entreprise>) (key, values) -> {
            Entreprise e = null;
            while (values.hasNext()) {
                Tuple2<Entreprise, Etablissement> tuple = values.next();
                if (e == null) {
                    e = tuple._1;
                }

                e.ajouterEtablissement(tuple._2);
            }

            return e;
        }, Encoders.bean(Entreprise.class))
                .foreach((ForeachFunction<Entreprise>) x -> System.out.println(x));

第二种解决方案这个解决方案更好，因为它可以优雅地处理企业中重复键的情况。它也将是高效的，因为它将减少要连接的记录数量。

KeyValueGroupedDataset<String, Etablissement> ets = dsEtablissements.groupByKey((MapFunction<Etablissement, String>) value -> value.siren, Encoders.STRING());

    Dataset<EtablissementList> etm = ets.mapGroups((MapGroupsFunction<String, Etablissement, EtablissementList>) (key, values) -> {
        Map<String, Etablissement> map = new HashMap<>();
        while (values.hasNext()) {
            Etablissement etablissement = values.next();
            map.put(etablissement.getId(), etablissement);
        }

        return new EtablissementList(map, key);
    }, Encoders.bean(EtablissementList.class));

    Dataset<Tuple2<Entreprise, EtablissementList>> dx = dsEntreprises.joinWith(etm, dsEntreprises.col("siren").equalTo(etm.col("siren")), "inner");
    Dataset<Entreprise> finalDs = dx.map((MapFunction<Tuple2<Entreprise, EtablissementList>, Entreprise>) value -> {
        value._1.etablissements = value._2.etablissements;
        return value._1;
    }, Encoders.bean(Entreprise.class));

    finalDs.foreach((ForeachFunction<Entreprise>) x -> System.out.println(x));

POJO

public static class EtablissementList {
        private Map<String, Etablissement> etablissements = new ConcurrentHashMap<>();
        private String siren;

        public EtablissementList() {
        }

        public EtablissementList(Map<String, Etablissement> etablissements, String siren) {
            this.etablissements = etablissements;
            this.siren = siren;
        }

        public Map<String, Etablissement> getEtablissements() {
            return etablissements;
        }

        public void setEtablissements(Map<String, Etablissement> etablissements) {
            this.etablissements = etablissements;
        }

        public String getSiren() {
            return siren;
        }

        public void setSiren(String siren) {
            this.siren = siren;
        }
    }