Spark-数据源及机器学习算法部署

Posted 2021-11-27 wind-man

1、数据源读取

使用的时候，需要加载驱动 --jars 或者添加到classpath中 或scaddjar

Spark对Oracle数据库读取，代码如下：

conf = SparkConf().setAppName(string_test)
sc = SparkContext(conf=conf)

ctx = SQLContext(sc)
sqltext = "(select dbms_lob.substr(title,500) as title,id,content,country,languages,time as publishDate,source,subject,source_url from news t where id <= 24) news"
news =ctx.read \
    .format("jdbc") \
    .option("url", "jdbc:oracle:thin:username/[email protected]//ip:port/sid") \
    .option("dbtable", sql) \
    .option("user", "user") \
    .option("password", "password") \
    .option("driver", "oracle.jdbc.driver.OracleDriver") \
    .load()

news.registerTempTable("news")

Spark 对Mongo读数据

ctx = SQLContext(sc)
mongourl = "mongodb://username:[email protected]:port"
mongoDB = "dbname"
mongoCollection = "collectionName"
mongoRows = ctx.read.format("com.mongodb.spark.sql").options(uri=mongourl,database=mongoDB, collection=mongoCollection).load()
mongoResultRdd = mongoRows.rdd

2、机器学习算法转换

机器学习算法有两类不能直接添加到spark中：

1) 包中含有复杂依赖关系的，如scipy、numpy等，scipy.special.beta函数在spark中不可以使用的。

2) 包不是.py结尾的，而是有第三方编译包的，不可以添加到spark中

 

解决办法：

在spark改写的代码中使用到上述相关的程序，阔以用subprocess调用python程序，以进行数据处理，然后得到程序返回结果。如下：

test= subprocess.getoutput("python /home/pytest.py \""+content.replace("\‘","’")+"\"")
re= test[test.index("::")+2:len(test)].replace(" ","")