我是齐柏林飞艇的新手,有些事情我只是不明白。
我已经使用python从数据库下载了一个表,然后,我想将其转换为RDD。但是我得到了找不到该表的错误。
我认为创建由其他解释器创建的表存在问题,但我真的不知道...我尝试了this和this问题,但仍然无法正常工作,它们直接用spark创建df。任何帮助都将非常有用:)
%python
engine = create_engine(
'mysql+mysqlconnector://...')
df = pd.read_sql(query, engine)
%spark
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext.implicits._
df=df.registerTempTable("df")
val df = sqlContext.sql("SELECT * from df LIMIT 5")
df.collect().foreach(println)
最佳答案
将Pandas DataFrame转换为Spark DataFrame非常简单:
%python
import pandas
pdf = pandas.DataFrame([[1, 2]]) # this is a dummy dataframe
# convert your pandas dataframe to a spark dataframe
df = sqlContext.createDataFrame(pdf)
# you can register the table to use it across interpreters
df.registerTempTable("df")
# you can get the underlying RDD without changing the interpreter
rdd = df.rdd
要使用scala spark获取它,您只需执行以下操作:
%spark
val df = sqlContext.sql("select * from df")
df.show()
// +---+---+
// | 0| 1|
// +---+---+
// | 1| 2|
// +---+---+
您还可以获取底层的rdd:
val rdd = df.rdd