【发布时间】:2021-03-12 23:33:00
【问题描述】:
我是 Apache Spark 框架的新手,我正在使用 ApacheSpark 通过 Hive 将数据写入 Hadoop。
在下面的代码中,我从 hive 中读取 table_1 并创建数据集,然后将此数据集映射到另一个数据集。毕竟将结果数据集保存到另一个 table_2。
public static void main(String[] args) {
Dataset<Row> ipSet = spark.sql("Select distinct ip from table_1");
ipSet.map(new TestFunction(), Encoders.bean(MyPojoClass.class)).write().mode("append").insertInto("table_2");
}
// ...
public class TestFunction implements MapFunction<Row, MyPojoClass>
{
private static List<..> staticData;
private static boolean callOnce = false;
private static void initStaticData() throws IOException
{
staticData = // set the data,
callOnce = true;
}
@Override
public MyPojoClass call(Row value) throws Exception {
if (!calledOnce)
initStaticData();
String ip = value.getAs("ip");
return new MyPojoClass(ip);
}
}
两个不同表中的行数应该相同,但是结果数据集/table_2中的行数不同。
例如,table_1 有 9.000.000 个不同的行,而 table_2 有 42.000 个不同的行。我该如何解决这个问题? (或者我错过了一些关于火花的概念?)
【问题讨论】:
-
您正在以不同的方式读取整个 table_1,因此您只在 table_2 中写入不同的值。同样在附加模式下,因此如果 table_2 中已经存在任何先前的值,则将被保留。他们都应该有相同的行数似乎很奇怪
-
@SCouto 9.000.000 是不同的数字。因此 table_2 必须有 9.000.000 个不同的行!
-
我对 Spark 和 Java 不太熟练,但对我来说似乎是对的。也许问题是表的元数据没有更新?执行插入后尝试在 table_2 上运行 MSCK 修复表,然后再次检查
标签: java apache-spark apache-spark-sql