我的目标是将输入文件读取到数据集中,将数据库中的数据读取到数据集中,将帐号上的数据集加入帐户,并将指定的信息输出到XML文件,每个文件有50K条记录。我总共有270万。 我能够成功读取文件,读取数据库,并将两者结合在一起。 finalDSExport.repartition(1).write().format("com.databricks.spark.xml").option("rootTag", "root").option("rowTag","record").save("test.xml"); 使用该代码,…

2020年1月12日 0条评论 27点热度 阅读全文

我想遍历spark DataFrame中列的内容,并在满足特定条件的情况下更正单元格中的数据 +-------------+ |column_title | +-------------+ +-----+ |null | +-----+ +-----+ |0 | +-----+ +-----+ |1 | +-----+ 假设我想在column的值为null时显示其他内容, Column.when() DataSet.withColumn() 但是我找不到正确的方法,我认为没有必要转换为RDD并对其进行迭代。 解决方…

2019年10月14日 0条评论 18点热度 阅读全文

我有一个仅包含标题(id,name,age)和0行的数据集。我想使用csv文件将其写入hdfs位置 DataFrameWriter dataFrameWriter = dataset.write(); Map<String, String> csvOptions = new HashMap<>(); csvOptions.put("header", "true"); dataFrameWriter = dataFrameWriter.options(csvOptions); dataFram…

2019年7月4日 0条评论 14点热度 阅读全文

有什么办法可以使用数据集解析多行json文件这是示例代码 public static void main(String[] args) { // creating spark session SparkSession spark = SparkSession.builder().appName("Java Spark SQL basic example") .config("spark.some.config.option", "some-value").getOrCreate(); Dataset<Row&…

2019年5月5日 0条评论 6点热度 阅读全文

假设我有一个数据集: Dataset<Row> sqlDF = this.spark.sql("SELECT first_name, last_name, age from persons"; 这将返回包含三列的 Dataset:first_name,last_name,age。 我想应用一个将 age列加5的函数,并返回一个与原始数据集具有相同列但年龄值已更改的新数据集: public int add_age(int old_age){ return old_age + 5; } 如何在Java上使用…

2019年3月5日 0条评论 15点热度 阅读全文

我正在使用Spark Java API,并尝试使用Dataset查找在2个文件之间删除的记录。对于我进行的一项测试,我正在比较具有2列的2个相同文件。我将其中一列用作PK的类型(如果PK不在较新的文件上,则表示删除)。 文件示例 ID | TYPE ABC |买入 CDE |购买 FGH |卖出 数据集创建为: Dataset<Row> previous/actual = sparkSession.read(). .option("inferSchema","true") .option("header…

2019年1月2日 0条评论 12点热度 阅读全文