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

HTTPX 基础教程-新乡seo|网站优化,网站建设_微信公众号:zeropython—昊天博客

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

spark.createDataFrame(rowRdd)

toDF() 转成 DataFrame

Centos 安装 Kafka 后台启动教程

Pyspark DataFrame 转成 rdd 互转

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

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

spark.createDataFrame(rowRdd)

toDF() 转成 DataFrame

Centos 安装 Kafka 后台启动 教程

Pyspark DataFrame 转成 rdd 互转

猜你喜欢

Linux 系统安装 Pyspark

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

Pyspark 从Kafka 接收数据

Pyspark DataFrame 转成 rdd 互转

Centos 安装 Kafka 后台启动 教程

Pyspark DataFrame 保存到MongoDB 数据库

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

关注我们的公众号

Centos 安装 Kafka 后台启动教程

Centos 安装 Kafka 后台启动教程

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