a.columns.toSet.intersect(b.columns.toSet).map(col).toSeq} val withoutHashDF = c.where(c("Hash_id").isNull)1
浏览 2提问于2017-02-24得票数 0
我有一个csv文件，我使用"SparkSession DataFrame“插入到insert的体验中。csv文件中的所有内容(头除外)都是整数。通过某些机器学习模型，dataframe按预期工作，直到尝试创建一个线性回归分类，在这里我得到了这个错误：
我相信这意味着数据不再是一个整
浏览 2提问于2017-12-18得票数 0
回答已采纳
在下面的代码中，我期望编译器识别output是在if部分或else部分中定义的。val df1 = spark.createDataFrame(Seq( (2, 20) val output = df1els
浏览 0提问于2019-01-18得票数 1
回答已采纳
在我的代码中，有一个dataframe是我从一个hive表创建的。我用函数过滤器做了一个简单的过滤器。但结果是随机的:我多次执行包含过滤器的同一个单元格，有时返回一行，有时不返回。我确信在预期的结果中只有一行。并且没有同时在源文件上的其他工作。df = spark.sql(&
浏览 0提问于2019-02-05得票数 0
我的应用程序是使用MongoDB作为平台构建的。DB中的一个集合具有大量的数据量，并且选择了apache通过计算来检索和生成分析数据。我已经将配置为与MongoDB通信。我需要使用MongoDB吡火花查询来查询集合，并构建一个由mongodb查询结果集组成的数据集。请给我一个适当的解决办法。
浏览 5提问于2016-08-09得票数 3
回答已采纳
1回答
我试着使用conv函数，就像我在一些例子中看到的那样，但并不适合我。我不明白为什么这个函数在DF列中返回相同的值。我在Scala2.11.11中使用了spark2.1，然后我尝试使用SPART2.2和Scala2.11.11。但是，当我使用应用于我的SHA2的conv函数时，它并不像预期的那样工作。我的代码是：
浏览 4提问于2018-06-10得票数 3
回答已采纳
我感到困惑的是，为什么在将得到的rdd.mapPartitions转换为DataFrame时，Spark似乎使用了1任务。所以我想知道为什么会这样，但从实际的角度来看，我主要关心的是能够从DataFrame --> mapParitions --到使用DataFrame.write过程。
这里是一个可重复的例子。下面的工作与预期一样，为mapPartitions工作<
浏览 4提问于2016-11-22得票数 17
回答已采纳
1回答
简单的请求是，我需要帮助将列添加到dataframe，但是，列必须是空的，它的类型来自...spark.sql.types，类型必须由字符串定义。不需要为org.apache.spark.sql.types中的每一种类型编写案例的东西df = df.withColumn("col_name", lit(null).cast(org.apa
浏览 4提问于2017-08-31得票数 2
回答已采纳
对于DLT来说，很难用python语法通过dlt.read_stream操作符作为另外两个活动表的联合(unionByName)返回数据。创建银表：最后，将更改应用于Silver视图创建为：def pre_merge_union_v:
df_table1 = <e
浏览 10提问于2022-10-08得票数 0
在我们的应用程序中，我们的大部分代码只是将filter、group by和aggregate操作应用于DataFrame，并将DF保存到Cassandra数据库。与下面的代码一样，我们有几个方法对不同数目的字段执行相同的filter, group by, join, agg操作，并返回一个DF，这些方法将保存到Cassandra表中。示例代码是：
val filteredDF = df.filter(col(&qu
浏览 5提问于2016-12-10得票数 0
1回答
我试过了 .parquet('*.parquet')但是，当我运行sql查询时，例如：它似乎只查看同一文件中的最后一个元素。这是否是一种预期的行为，如
浏览 0提问于2019-07-16得票数 2
我正在尝试将DF保存为s3上的json格式。它被保存为json对象文件，但是我想要json数组文件。s3, which i am loading into dataframe in aws glue. after performing some transformation i am writing DFapplymapping1") applymapping2 = applymapping1.toDF() applymapping2.coalesce(1).write.format("org.apache.sp
浏览 8提问于2019-04-23得票数 0
我使用PySpark将几个文件读入数据帧，并执行它们的联合。但是，当我尝试读取第二个文件(我可以单独读取这两个文件中的任何一个，但不能一起读取)时，我得到一个错误。= spark.read.parquet(file_path)
return df 现在，下面的代码在两个不同的pyspark-session中工作： df1 = read_file(fileprint(df3.
浏览 16提问于2020-11-02得票数 0
1回答
自从上次使用spark太久了，我再次使用Spark3.1，下面是我的问题:我还有2000万行加入400米行，原始代码是：for time当查看星火历史服务器上的可视化SQL图时，我的改进解决方案(?)在第二个联接中，它在每次迭代时再次使用整个左联接，而不是使用更干净、更轻的DataFrame。我的最后一个想法
浏览 1提问于2021-08-28得票数 1
作为输入，我有一个包含int值的csv文件。我还设计了一些用于numpy数组输入的python函数，这些函数需要应用于星火DataFrame。().sum().collect()
但是，我需要的是一个标准<em
浏览 4提问于2021-02-08得票数 2
回答已采纳
当我试图从empty_df中“减去”original_df并计算这两个数据时，结果是不同的： StructField("guid", StringType(), True)original_d
浏览 1提问于2019-01-24得票数 0
回答已采纳
我有两个数据- target_df和reference_df.我需要删除target_df中的account_id，它存在于reference_df中。target_df是从hive表创建的，将有数百个分区。它是根据日期(20220101到20221101)进行分区的。
我是做左反连接和写入数据在hdfs的位置。val numPartitions
浏览 11提问于2022-11-21得票数 0
1回答
No such file or directory 's3://<myPrefix>/part-00001-a123a120-7d11-581a-b9df-bc53076d57894-c000.snappy.parquet如果我尝试写到一个新的S3位置，比如s3://dataset_new_path.../，那么代码就能正常工作。my_df \ .format(
浏览 7提问于2019-10-13得票数 1
回答已采纳
我正在尝试做的是添加一个新的列(称为'churn')，并根据它是否是该特定客户作为客户的最后一个月将该列设置为0或1。我已经尝试了许多方法来做到这一点，但每一个都失败了，要么是回溯，要么就是不能像预期的那样工作。应该注意的是，我对python和pandas都是新手，所以请解释一下像我五岁(笑)这样的东西。max(date) else 0 for date in
浏览 5提问于2021-06-18得票数 1
3回答
我有一些过滤器，我需要应用到星火中的数据帧，但首先在运行时，我知道哪些过滤器给用户。目前，我正在将它们添加到单个filter函数中，但如果没有定义其中一个文件，则会失败。我是否应该通过创建一个大过滤器来做到这一点：if (_filter2 !或者我还没有在星火中找到一个很好的模式呢？
浏览 4提问于2017-03-13得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
