pyspark-sql

1热度

1回答

我有被注册为tempView一个数据帧和蜂巢表加入 df1.createOrReplaceTempView("mydata") df2 = spark.sql("Select md.column1,md.column2,mht.column1 \ from mydata md inner join myHivetable mht on mht.key1 = md.ke

0热度

1回答

pyspark使用sparkcontext</p> <p>我的数据库远程机器

我使用Python 2.7使用Ubuntu和运行火花使用mysql数据库通过Python脚本是一个远程MySQL，使用用户名和密码。我尝试使用此代码 sc = createSparkContext() sql = SQLContext(sc) df = sql.read.format('jdbc').options(url='jdbc:mysql://ip:port?user=user&pa

0热度

1回答

如何检查Pyspark地图中是否存在关键字或值

我在Spark DF中有一个Map列，并且想要在特定键上过滤此列（即，如果地图中的键与所需值匹配，则保留该列）。例如，我的模式被定义为： df_schema = StructType( [StructField('id', StringType()), StructField('rank', MapType(StringType(), IntegerType()))] )

0热度

1回答

spark 2.x正在读取整数/双列作为使用csv函数的字符串

我正在使用下面的语句读取spark中的csv。 df = spark.read.csv('<CSV FILE>', header=True, inferSchema = True) 当我在火花数据帧正在检查一些整数和双列的形式存储在数据框中字符串列。但是，所有列都不是这种情况。我已经检查了特定列的值，所有的值都是double类型，但仍然是spark，推断为StringType。由于我使用大

0热度

1回答

火花2.0.0选择不同的结果不稳定

我跑pyspark2过火花2.0.0 考虑加载到数据帧稳定常数的数据集，我也被某些列拖放重复： vw_ticket = read_csv(...) vw_ticket_in_booking = vw_ticket.dropDuplicates(['BOOKING_TICKET_ID']) vw_ticket_in_booking.createOrReplaceTempView('vw_tick

0热度

2回答

如何在pyspark中导入'þ'分隔的.txt文件

我在AWS s3中有一个分隔的.txt文件。 datan用þ839729þ25þ[email protected]þfirstnameþlastnameþ0þBLACKþ28/08/2017þ12329038þ99þ287þ81þ0 我试着用databricks and sparkcontext来导入数据。虽然databricks方法运行并没有抛出错误，但数据框中没有数据。火花背景只是把错误说 -

1热度

3回答

如何从Kafka读取XML格式的流数据？

我正在尝试使用Spark结构化流从卡夫卡主题读取XML数据。我试过使用Databricks spark-xml包，但是我收到一个错误消息，说这个包不支持流式阅读。有什么方法可以使用结构化流从Kafka主题中提取XML数据？我当前的代码： df = spark \ .readStream \ .format("kafka") \ .format('com.da

0热度

1回答

聚合和合并RDD的正确方法

我有一个客户表，其中包含有关每个客户的多个进程的信息。目标是为每个客户和每个过程提取功能。这意味着每个特征主要是对一个对象的聚合或分类比较计算。但是，目标是能够随着时间的推移添加越来越多的功能。所以基本上用户应该能够使用一些过滤器，度量和聚合来定义一个新的函数，并将这个新函数添加到在表上操作的函数池中。输出应该是具有所有功能的customerID，processID表。所以我startet

0热度

1回答

MySQL的阅读PySpark

我有以下的测试代码： from pyspark import SparkContext, SQLContext sc = SparkContext('local') sqlContext = SQLContext(sc) print('Created spark context!') if __name__ == '__main__': df = sqlContext.rea

0热度

1回答

在pyspark中加载SQl查询？

有没有什么方法可以直接从oracle sql使用pyspark来读取sql脚本？我一直在pyspark中加载两张桌子并试图加入，在pyspark花费很长时间，我需要加入更多桌子（6张桌子）。这在Oracle Sql中并不需要太多时间。那么我怎样才能读取和运行一个“X.sql”与pyspark文件？