Spark 1.6.2中Scala环境下RDD多字段混合升降序排序方法

Got it, let's work through how to get your sorted product RDD in Spark 1.6.2 with Scala 2.10.5. Here's a step-by-step breakdown:

First, you'll need the MySQL JDBC driver to connect Spark to your InnoDB table. Since you're using Spark 1.6.2 (an older version), stick with a compatible MySQL driver like mysql-connector-java-5.1.47.jar (newer 8.x drivers might have compatibility issues).

When submitting your Spark job, include the driver with the --jars flag:

spark-submit --class com.yourpackage.ProductSorting --jars mysql-connector-java-5.1.47.jar your-assembled-jar.jar

We'll read the products table, convert rows to a type-safe case class, then apply the required sorting.

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.Row

// Define a case class to map the products table structure (type-safe and easier to work with)
case class Product(
  product_id: Int,
  product_category_id: Int,
  product_name: String,
  product_description: String,
  product_price: Float,
  product_image: String
)

object ProductSorting {
  def main(args: Array[String]): Unit = {
    // Initialize Spark configuration and context
    val conf = new SparkConf()
      .setAppName("ProductSortingJob")
      .setMaster("local[*]") // Replace with your cluster master URL in production
    val sc = new SparkContext(conf)

    // JDBC connection parameters
    val jdbcUrl = "jdbc:mysql://your-db-host:3306/your-db-name?useUnicode=true&characterEncoding=utf8"
    val dbProps = new java.util.Properties()
    dbProps.setProperty("user", "your-db-username")
    dbProps.setProperty("password", "your-db-password")
    dbProps.setProperty("driver", "com.mysql.jdbc.Driver")

    // Read products table into an RDD[Row]
    val rawProductsRDD: RDD[Row] = sc.jdbc(jdbcUrl, "products", dbProps)

    // Convert raw RDD[Row] to RDD[Product] for easier manipulation
    val productsRDD: RDD[Product] = rawProductsRDD.map(row => Product(
      row.getInt(0),          // product_id
      row.getInt(1),          // product_category_id
      row.getString(2),       // product_name
      row.getString(3),       // product_description
      row.getFloat(4),        // product_price
      row.getString(5)        // product_image
    ))

    // Sort by product_name (ascending) first, then product_price (descending)
    // We use (-product_price) to reverse the sort order for price while keeping name in ascending order
    val sortedProductsRDD: RDD[Product] = productsRDD.sortBy(product => (product.product_name, -product.product_price), ascending = true)

    // Example: Print the sorted results (replace with saveAsTextFile or other output as needed)
    sortedProductsRDD.foreach(product => println(s"${product.product_name} | Price: ${product.product_price} | ID: ${product.product_id}"))

    // Clean up Spark context
    sc.stop()
  }
}

• Type Safety: Using a case class makes your code more readable and avoids errors from accessing row indices incorrectly.
• Sort Logic: The sortBy function uses tuple comparison: it first sorts by product_name (ascending), then by -product_price (ascending on the negative value equals descending on the original price).
• Encoding: The JDBC URL includes useUnicode=true&characterEncoding=utf8 to match your table's utf8 charset and prevent garbled text.
• Performance: For large datasets, add partitioning parameters to the sc.jdbc call (like partitionColumn, lowerBound, upperBound, numPartitions) to parallelize the read and speed up processing.

内容的提问来源于stack exchange，提问作者Choix