pyspark的rdd直接写入mysql

Google搜索"RDD write into mysql"前面5页得到:[5][6][7][8][9][10]

我们一个个来分析

[1][2][3]读出的是RDD,写入的是foreachpartition的方式

[4]写入的不是spark RDD,而是一个Spark的DataFrame类型的变量

[5]写入的不是spark RDD,而是Spark RDD转化为DataFrame类型然后写入

[6][7]写入的是spark DataFrame

[8]仅仅是代码bug而已

[9]写入的是dataframe不是rdd

[10]提到了检查partition的数量可以加速rdd的写入，里面有两种方式，第一种是RDD写入，第二种是Data Frame写入。

结论:

[1][2][3][10]有用(都是scala语言)，其他都没有用。

[11]提到了针对partition来提高速度。

尝试Pyspark版本:

import pandas as pd
from pyspark.sql import SparkSession
from pyspark import SparkContext
from pyspark.sql import SQLContext
from pymysql import *def map_extract(element):file_path, content = elementyear = file_path[-8:-4]return [(year, i) for i in content.split("\n") if i]spark = SparkSession\.builder\.appName("PythonTest")\.getOrCreate()res = spark.sparkContext.wholeTextFiles('hdfs://Desktop:9000/user/mercury/names',minPartitions=40)  \.map(map_extract) \.flatMap(lambda x: x) \.map(lambda x: (x[0], int(x[1].split(',')[2]))) \.reduceByKey(lambda x,y:x+y)def write2mysql(x):conn = connect(host='Desktop', port=3306, database='leaf', user='appleyuchi',password='appleyuchi', charset='utf8')cs1 = conn.cursor()
# 执行sql语句
#---------------------------------------for item in x:key = str(item[0])num = item[1]# values = (key,num)# print("values=",values)sql = "insert into `spark`(`key`,`num`) values (%s,%s)"cs1.execute(sql,(key,num))# 提交之前的操作，如果之前已经执行多次的execute，那么就都进行提交conn.commit()
#---------------------------------------# 关闭cursor对象cs1.close()# 关闭connection对象conn.close()res.foreachPartition(lambda x:write2mysql(x))

数据集用的是：

https://github.com/wesm/pydata-book/tree/2nd-edition/datasets/babynames

如果报错:

pymysql.err.InternalError: (1049, "Unknown database 'leaf'")

查过存在该数据库还是解决不了的话，参考:

https://gitee.com/appleyuchi/cluster_configuration/blob/master/物理环境配置流程-必须先看.txt

Reference:

[1]RDD从mysql中读取数据和RDD往数据库中存数据(代码完整)

[2]RDD 直接存入MySQL，以及直接读取MySQL中数据(代码不完整)

[3]spark以rdd方式读写mysql

[4]Spark RDD写入RMDB(Mysql)方法二

[5]How to put data from Spark RDD to Mysql Table

[6]WRITING TO A DATABASE FROM SPARK

[7]save spark rdd into Mysql

[8]Writing from PySpark to MySQL Database

[9]spark 2.1写入mysql spark 2.1 write to mysql

[10]how to properly save spark rdd result to a mysql database(代码不完整)

[11]计算质数通过分区(Partition)提高Spark的运行性能

pyspark的rdd直接写入mysql相关推荐

pyspark.sql写入mysql_将pyspark dataframe写入MySQL数据库时出错
我得到以下错误: "Caused by: java.lang.NoSuchMethodException: org.apache.spark.sql.execution.datasource ...
scala写入mysql_spark rdd转dataframe 写入mysql的实例讲解
dataframe是在spark1.3.0中推出的新的api,这让spark具备了处理大规模结构化数据的能力,在比原有的RDD转化方式易用的前提下,据说计算性能更还快了两倍.spark在离线批处理或者 ...
SparkStreaming读取Kafka数据源并写入Mysql数据库
SparkStreaming读取Kafka数据源并写入Mysql数据库一.实验环境本实验所用到的工具有 kafka_2.11-0.11.0.2: zookeeper-3.4.5: spark-2. ...
114.Spark大型电商项目-广告点击流量实时统计-使用高性能方式将实时计算结果写入MySQL中
目录误区 Spark Streaming foreachRDD的正确使用方式对于这种实时计算程序的mysql插入,有两种pattern(模式) 代码 AdUserClickCount.java I ...
pandas读取大文件（chunksize）并通过sqlalchemy写入MySQL数据库
pandas读取大文件(chunksize)并通过sqlalchemy写入MySQL数据库在pandas中读取表类文件的时候有一个参数chunksize,只要指定了这个参数的数值,那么得到的结果就不 ...
python爬取新闻并归数据库_Python爬取数据并写入MySQL数据库操作示例
Python爬取数据并写入MySQL数据库的实例首先我们来爬取 http://html-color-codes.info/color-names/ 的一些数据. 按 F12 或 ctrl+u 审查元 ...
关于log4net日志写入mysql数据库记录
网上关于log4net日志写入mysql数据库的博客感觉比较少,所以这边搞定之后先过来记录一下. 首先新建个项目,我命名是log4netDemo,然后需要引入两个dll,一个是mysql.dll,一个 ...
python爬取mysql数据_Python爬取数据并写入MySQL数据库的实例
Python爬取数据并写入MySQL数据库的实例来源:中文源码网浏览: 次日期:2018年9月2日 [下载文档: Python爬取数据并写入MySQL数据库的实例.txt ] (友 ...
写入MySQL报错超出 max_allowed_packet 的问题
写入MySQL报错超出 max_allowed_packet 的问题. MySQL会根据配置文件会限制server接受的数据包的大小.如果写入大数据时,因为默认的配置太小,插入和更新操作会因为 max ...

pyspark的rdd直接写入mysql

pyspark的rdd直接写入mysql相关推荐

最新文章

热门文章