Pyspark 从kafka 读取数据 rdd，转成 DataFrame-昊天SEO

spark.createDataFrame(rowRdd)

from __future__ import print_function

import sys
import json
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from pyspark.sql import Row, SparkSession

def getSparkSessionInstance(sparkConf):
     # 生成单例模式的getSparkSessionInstance
    if ('sparkSessionSingletonInstance' not in globals()):
        globals()['sparkSessionSingletonInstance'] = SparkSession
            .builder
            .config(conf=sparkConf)
            .getOrCreate()
    return globals()['sparkSessionSingletonInstance']




sc = SparkContext("local[2]","NetWordCount")
ssc = StreamingContext(sc,1)

topic  = "connect-test"
# kafka 中的 topic

kvs = KafkaUtils.createStream(ssc,"localhost:2181","spark-streaming-consumer",{topic:1})
# kafka 读取数据

words = kvs.map(lambda x:x[1])
# words = kvs.map(lambda line: line.split(","))
# words = kvs.flatMap(lambda line: line.split(" "))

# Convert RDDs of the words DStream to DataFrame and run SQL query


def process(time, rdd):
    print("========= %s =========" % str(time))

    try:
        # Get the singleton instance of SparkSession
        spark = getSparkSessionInstance(rdd.context.getConf())


        # Convert RDD[String] to RDD[Row] to DataFrame
        rowSplitted = rdd.flatMap(lambda line: line.split(","))
        rowRdd = rowSplitted.map(lambda w: Row(word=w))
        # rowRdd = rdd.flatMap(lambda line: line.split(","))
        wordsDataFrame = spark.createDataFrame(rowRdd)

        # Creates a temporary view using the DataFrame.
        wordsDataFrame.createOrReplaceTempView("words")

        # Do word count on table using SQL and print it
        wordCountsDataFrame = 
            spark.sql("select word, count(*) as total from words group by word")
        wordCountsDataFrame.show()

    except:
        pass

words.pprint()
words.foreachRDD(process)

# foreachRDD Apply a function to each RDD in this DStream(流处理).

ssc.start()
ssc.awaitTermination()

from __future__ import print_function

import sys

import json

from pyspark import SparkContext

from pyspark.streaming import StreamingContext

from pyspark.streaming.kafka import KafkaUtils

from pyspark.sql import Row, SparkSession

def getSparkSessionInstance(sparkConf):

# 生成单例模式的getSparkSessionInstance

if ('sparkSessionSingletonInstance' not in globals()):

globals()['sparkSessionSingletonInstance'] = SparkSession

.builder

.config(conf=sparkConf)

.getOrCreate()

return globals()['sparkSessionSingletonInstance']

sc = SparkContext("local[2]","NetWordCount")

ssc = StreamingContext(sc,1)

topic = "connect-test"

# kafka 中的 topic

kvs = KafkaUtils.createStream(ssc,"localhost:2181","spark-streaming-consumer",{topic:1})

# kafka 读取数据

words = kvs.map(lambda x:x[1])

# words = kvs.map(lambda line: line.split(","))

# words = kvs.flatMap(lambda line: line.split(" "))

# Convert RDDs of the words DStream to DataFrame and run SQL query

def process(time, rdd):

print("========= %s =========" % str(time))

try:

# Get the singleton instance of SparkSession

spark = getSparkSessionInstance(rdd.context.getConf())

# Convert RDD[String] to RDD[Row] to DataFrame

rowSplitted = rdd.flatMap(lambda line: line.split(","))

rowRdd = rowSplitted.map(lambda w: Row(word=w))

# rowRdd = rdd.flatMap(lambda line: line.split(","))

wordsDataFrame = spark.createDataFrame(rowRdd)

# Creates a temporary view using the DataFrame.

wordsDataFrame.createOrReplaceTempView("words")

# Do word count on table using SQL and print it

wordCountsDataFrame =

spark.sql("select word, count(*) as total from words group by word")

wordCountsDataFrame.show()

except:

pass

words.pprint()

words.foreachRDD(process)

# foreachRDD Apply a function to each RDD in this DStream(流处理).

ssc.start()

ssc.awaitTermination()

toDF() 转成 DataFrame

>>> from pyspark.sql.types import Row  
>>> def f(x):
... rel = {}  
... rel['name'] = x[0]  
... rel['age'] = x[1]  
... return rel  
...  
>>> peopleDF = sc.textFile("file:///usr/local/spark/examples/src/main/resources/people.txt").map(lambda line : line.split(',')).map(lambda x: Row(**f(x))).toDF()  
>>> peopleDF.createOrReplaceTempView("people") //必须注册为临时表才能供下面的查询使用

>>> personsDF = spark.sql("select * from people")  
>>> personsDF.rdd.map(lambda t : "Name:"+t[0]+","+"Age:"+t[1]).foreach(print)
Name: 19,Age:Justin  
Name: 29,Age:Michael  
Name: 30,Age:Andy

>>> from pyspark.sql.types import Row

>>> def f(x):

... rel = {}

... rel['name'] = x[0]

... rel['age'] = x[1]

... return rel

...

>>> peopleDF = sc.textFile("file:///usr/local/spark/examples/src/main/resources/people.txt").map(lambda line : line.split(',')).map(lambda x: Row(**f(x))).toDF()

>>> peopleDF.createOrReplaceTempView("people") //必须注册为临时表才能供下面的查询使用

>>> personsDF = spark.sql("select * from people")

>>> personsDF.rdd.map(lambda t : "Name:"+t[0]+","+"Age:"+t[1]).foreach(print)

Name: 19,Age:Justin

Name: 29,Age:Michael

Name: 30,Age:Andy

Pyspark 从kafka 读取数据 rdd，转成 DataFrame

spark.createDataFrame(rowRdd)

toDF() 转成 DataFrame

Centos 安装 Kafka 后台启动教程

Pyspark DataFrame 转成 rdd 互转

发表评论点击这里取消回复。

关注微信公众号有惊喜么么哒！

最热点击文章

使用Spark 登录 QQ 邮箱

Django 多对多正向查询反向查询

PHPSTORM Laravel 配置基础教程

昊园恒业黑中介，谁来惩罚

Django开发常用插件

使用 ffmpeg 转换视频格式 mp4 webm

Django 多对多的保存

Python 批量向Django 导入数据

uwsgi重启shell脚本

Django 发送html格式的邮件

最新评论

Pyspark 从kafka 读取数据 rdd，转成 DataFrame

spark.createDataFrame(rowRdd)

toDF() 转成 DataFrame

Centos 安装 Kafka 后台启动 教程

Pyspark DataFrame 转成 rdd 互转

猜你喜欢

Kafka.errors.NoBrokersAvailable: NoBrokersAvailable. [亲测]

Pyspark 从Kafka 接收数据

Pyspark DataFrame 转成 rdd 互转

Centos 安装 Kafka 后台启动 教程

Pyspark DataFrame 保存到MongoDB 数据库

Python Spark 链接 MongoDB

发表评论 点击这里取消回复。

关注微信公众号有惊喜 么么哒！

最热点击文章

使用Spark 登录 QQ 邮箱

Django 多对多 正向查询 反向查询

PHPSTORM Laravel 配置 基础教程

昊园恒业 黑中介，谁来惩罚

Django开发常用插件

使用 ffmpeg 转换视频格式 mp4 webm

Django 多对多的保存

Python 批量向Django 导入数据

uwsgi重启shell脚本

Django 发送html格式的邮件

最新评论

关注我们的公众号

Centos 安装 Kafka 后台启动教程

Centos 安装 Kafka 后台启动教程

发表评论点击这里取消回复。

关注微信公众号有惊喜么么哒！

Django 多对多正向查询反向查询

PHPSTORM Laravel 配置基础教程

昊园恒业黑中介，谁来惩罚