更新时间:2022-05-25 GMT+08:00
分享

pyspark样例代码

开发说明

mongo只支持增强型跨源。只能使用包年包月队列。

  • 前提条件

    在DLI管理控制台上已完成创建增强跨源连接,并绑定包年/包月队列。具体操作请参考《数据湖探索用户指南》。

  • 通过DataFrame API 访问
    1. import相关依赖
      from __future__ import print_function
      from pyspark.sql.types import StructType, StructField, IntegerType, StringType
      from pyspark.sql import SparkSession
    2. 创建session
      1
      sparkSession = SparkSession.builder.appName("datasource-mongo").getOrCreate()
      
    3. 设置连接参数
      1
      2
      3
      4
      5
      6
      url = "192.168.4.62:8635,192.168.5.134:8635/test?authSource=admin"
      uri = "mongodb://username:pwd@host:8635/db"
      user = "rwuser"
      database = "test"
      collection = "test"
      password = "######"
      

      详细的参数说明请参考表1

    4. 创建DataFrame
      1
      2
      3
      4
      5
      dataList = sparkSession.sparkContext.parallelize([("1", "Katie", 19),("2","Tom",20)])
      schema = StructType([StructField("id", IntegerType(), False),          
                           StructField("name", StringType(), False),
                           StructField("age", IntegerType(), False)])
      dataFrame = sparkSession.createDataFrame(dataList, schema)
      
    5. 导入数据到mongo
      1
      2
      3
      4
      5
      6
      7
      8
      9
      dataFrame.write.format("mongo")
        .option("url", url)
        .option("uri", uri)
        .option("user",user)
        .option("password",password)
        .option("database",database)
        .option("collection",collection)
        .mode("Overwrite")
        .save()
      
    6. 读取Mongo上的数据
      jdbcDF = sparkSession.read
        .format("mongo")
        .option("url", url)
        .option("uri", uri)
        .option("user",user)
        .option("password",password)
        .option("database",database)
        .option("collection",collection)
        .load()
      jdbcDF.show()
    7. 操作结果

  • 通过SQL API 访问
    1. 创建DLI关联跨源访问 Mongo的关联表。
      1
      2
      3
      4
      5
      6
      7
      8
      sparkSession.sql(
          "create table test_mongo(id string, name string, age int) using mongo options(
            'url' = '192.168.4.62:8635,192.168.5.134:8635/test?authSource=admin',
            'uri' = 'mongodb://username:pwd@host:8635/db',
            'database' = 'test',
            'collection' = 'test', 
            'user' = 'rwuser', 
            'password' = '######')")
      

      详细的参数说明请参考表1

    2. 插入数据
      1
      sparkSession.sql("insert into test_mongo values('3', 'zhangsan',23)")
      
    3. 查询数据
      1
      sparkSession.sql("select * from test_mongo").show()
      
  • 提交Spark作业
    1. 将写好的python代码文件上传至DLI中。控制台操作请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>《上传资源包》。
    2. 在Spark作业编辑器中选择对应的Module模块并执行Spark作业。控制台操作请参考《数据湖探索用户指南》。API操作请参考《数据湖探索API参考》>《创建批处理作业》。
      • 提交作业时,需要指定Module模块,名称为:sys.datasource.mongo。
      • 通过控制台提交作业请参考《数据湖探索用户指南》中的“选择依赖资源参数说明”表说明。
      • 通过API提交作业请参考《数据湖探索API参考》>《创建批处理作业》中“表2-请求参数说明”关于“modules”参数的说明。

完整示例代码

  • 通过DataFrame API 访问
    from __future__ import print_function
    from pyspark.sql.types import StructType, StructField, IntegerType, StringType
    from pyspark.sql import SparkSession
    
    if __name__ == "__main__":
      # Create a SparkSession session.
      sparkSession = SparkSession.builder.appName("datasource-mongo").getOrCreate()
    
      # Create a DataFrame and initialize the DataFrame data.  
      dataList = sparkSession.sparkContext.parallelize([("1", "Katie", 19),("2","Tom",20)])
      
      # Setting schema  
      schema = StructType([StructField("id", IntegerType(), False),StructField("name", StringType(), False), StructField("age", IntegerType(), False)])
    
      # Create a DataFrame from RDD and schema  
      dataFrame = sparkSession.createDataFrame(dataList, schema)
    
      # Setting connection parameters  
      url = "192.168.4.62:8635,192.168.5.134:8635/test?authSource=admin"
      uri = "mongodb://username:pwd@host:8635/db"
      user = "rwuser"
      database = "test"
      collection = "test"
      password = "######"
     
      # Write data to the mongodb table  
      dataFrame.write.format("mongo")
        .option("url", url)
        .option("uri", uri)
        .option("user",user)
        .option("password",password)
        .option("database",database)
        .option("collection",collection)
        .mode("Overwrite").save()
    
      # Read data  
      jdbcDF = sparkSession.read.format("mongo")
        .option("url", url)
        .option("uri", uri)
        .option("user",user)
        .option("password",password)
        .option("database",database)
        .option("collection",collection)
        .load()   
      jdbcDF.show()
     
      # close session  
      sparkSession.stop()
  • 通过SQL API 访问
    from __future__ import print_function
    from pyspark.sql import SparkSession
    
    if __name__ == "__main__":
      # Create a SparkSession session.  
      sparkSession = SparkSession.builder.appName("datasource-mongo").getOrCreate()
    
      # Createa data table for DLI - associated mongo
      sparkSession.sql(
        "create table test_mongo(id string, name string, age int) using mongo options(
          'url' = '192.168.4.62:8635,192.168.5.134:8635/test?authSource=admin',
          'uri' = 'mongodb://username:pwd@host:8635/db',
          'database' = 'test',
          'collection' = 'test', 
          'user' = 'rwuser', 
          'password' = '######')")
    
      # Insert data into the DLI-table  
      sparkSession.sql("insert into test_mongo values('3', 'zhangsan',23)")
    
      # Read data from DLI-table  
      sparkSession.sql("select * from test_mongo").show()
    
      # close session  
      sparkSession.stop()
分享:

    相关文档

    相关产品

close