6.Best Practices for Handling Big Data with Python in

    import org.apache.hadoop.io.{LongWritable, Text}
    import org.apache.hadoop.mapreduce.Mapper
    import org.apache.hadoop.mapreduce.Reducer
    
    class Tokenizer extends Mapper[Object, Text, Text, LongWritable] {
      def map(key: Object, value: Text, context: Context): Unit = {
    val line = value.toString()
    for (token <- line.split("\ W+")) {
      if (token!= "") {
        context.write(new Text(token), new LongWritable(1))
      }
    }
      }
    }
    
    class SumReducer extends Reducer[Text, LongWritable, Text, LongWritable] {
      def reduce(key: Text, values: java.lang.Iterable[LongWritable],
             context: Reducer[Text, LongWritable, Text, LongWritable]#Context): Unit = {
    var sum = 0L
    for (value <- values) {
      sum += value.get()
    }
    context.write(key, new LongWritable(sum))
      }
    }
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    val wordCounts = sc.textFile("s3a://mybucket/output")
                     .flatMap(_.split("\t"))
                     .filter(_!= " ").map{case x => (x.substring(0, x.length-1), x.substring(x.length).toLong)}
    println(wordCounts.collect().mkString(", "))
    
      
      
      
    
    代码解读

    // 将数据集保存为parquet文件
    df.write.mode("overwrite").format("parquet").save("s3a://mybucket/path/to/output/")
    
    // 从parquet文件中读取数据集
    val df = spark.read.format("parquet").load("s3a://mybucket/path/to/input/")
    
    // 对数据集进行过滤、聚合、排序
    val filteredDf = df.filter($"age" > 10).groupBy($"gender").agg(sum($"amount")).orderBy($"sum(amount)" desc)
    
      
      
      
      
      
      
      
    
    代码解读

    import org.apache.kafka.clients.consumer._
    import org.apache.kafka.common.serialization.StringDeserializer
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
    
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "test",
      "auto.offset.reset" -> "earliest"
    )
    
    val topics = Array("topic1", "topic2")
    
    val ssc = new StreamingContext(spark.sparkContext, Seconds(1))
    
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaParams))
    
    stream.foreachRDD((rdd, time) => {
      // perform analysis on the RDD here
      println(s"========= $time =========")
      rdd.foreach(record => println(record.value()))
    })
    
    ssc.start()
    ssc.awaitTermination()
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    SELECT * FROM hive.default.orders WHERE order_date >= DATEADD('day', -7, GETDATE())
    
    
    代码解读

    CREATE TABLE IF NOT EXISTS orders (order_id INT, customer VARCHAR, product VARCHAR, price DECIMAL(10, 2), quantity INT) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',';
    LOAD DATA INPATH's3n://mybucket/path/to/input' OVERWRITE INTO TABLE orders;
    
    SELECT customer, AVG(price*quantity) AS total_revenue
    FROM orders
    WHERE order_date >= DATEADD('day', -7, GETDATE())
    GROUP BY customer;
    
      
      
      
      
      
      
    
    代码解读

    sudo yum update -y && sudo yum install git python3 -y
    git clone https://github.com/apache/spark.git /opt/spark
    cd /opt/spark
    ./build/mvn -DskipTests clean package
    export PYSPARK_PYTHON=/usr/bin/python3
    ./bin/pyspark --packages org.apache.spark:spark-avro_${SCALA_VERSION}:${SPARK_VERSION} \
              --conf "spark.jars.ivy=/root/.ivy2" \
              --master local[*] --driver-memory=1g --executor-memory=1g \
              --num-executors 2 --executor-cores 2
    
      
      
      
      
      
      
      
      
    
    代码解读

    from pyspark import SparkConf, SparkContext
    
    conf = SparkConf().setAppName("TestApp").setMaster("yarn")\
                 .set("spark.executor.instances", "2")\
                 .set("spark.executor.memoryOverhead", "1g")\
                 .set("spark.executor.cores", "2")\
                 .set("spark.driver.cores", "1")\
                 .set("spark.dynamicAllocation.enabled", "true")\
                 .set("spark.shuffle.service.enabled", "true")
    sc = SparkContext(conf=conf)
    
    textData = sc.textFile("/mnt/data/example.txt")
    print(textData.first()) # output first line of text file
    
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    import boto3
    
    s3 = boto3.resource('s3')
    
    s3.meta.client.upload_file('/tmp/example.csv', 'your-bucket', 'path/to/example.csv')
    
    object = s3.Bucket('your-bucket').Object('path/to/example.csv').download_file('/tmp/downloaded-example.csv')
    
      
      
      
      
      
      
    
    代码解读

    AWSTemplateFormatVersion: 2010-09-09
    Transform: AWS::Serverless-2016-10-31
    Description: Example Serverless Application
    Resources:
      MyFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: index.handler
      Runtime: nodejs10.x
      CodeUri: functions/
      Description: An example Lambda function that returns a message
      MemorySize: 128
      Timeout: 3
      Events:
        ApiEvent:
          Type: Api
          Properties:
            Path: '/hello'
            Method: get
      ApiGatewayRestApi:
    Type: AWS::ApiGateway::RestApi
    Properties:
      Body: '{"swagger": "2.0","info":{"title":"My API","version":"1.0"}}'
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    exports.handler = async (event, context) => {
      const response = {
    statusCode: 200,
    body: JSON.stringify({message: 'Hello world!'})
      };
    
      return response;
    };
    
      
      
      
      
      
      
      
    
    代码解读

    import requests
    response = requests.get('<api endpoint>/hello')
    if response.status_code == 200:
      print(response.json()['message'])
    else:
      print('Error:', response.text)
    
      
      
      
      
      
    
    代码解读