正文

sparkR的一个运行的例子

IP云
IP云V管理员/0 阅读
10/02
小程序：扫一扫查出行
【扫一扫了解最新限行尾号】
复制小程序

在sparkR在配置完成的基础上，本例采用Spark on yarn模式，介绍sparkR运行的一个例子。

在spark的安装目录下，/examples/src/main/r，有一个dataframe.R文件。该文件默认是在本地的模式下运行的，不与hdfs交互。可以将脚本进行相应修改，提交到yarn模式下。

在提交之前，要先将${SPARK_HOME}/examples/src/main/resources/people.json 文件上传到hdfs上，我上传到了hdfs://data-mining-cluster/data/bigdata_mining/sparkr/people.json 目录下。

  1. dataframe.R文件内容：
  1. # Convert local data frame to a SparkR DataFrame
  2. df <- createDataFrame(sqlContext, localDF)
  3. # Print its schema
  4. printSchema(df)
  5. # root
  6. #  |-- name: string (nullable = true)
  7. #  |-- age: double (nullable = true)
  8. # Create a DataFrame from a JSON file
  9. #path <- file.path("hdfs://data-mining-cluster/data/bigdata_mining/sparkr/people.json")
  10. path <- file.path("hdfs://data-mining-cluster/data/bigdata_mining/sparkr/people.json")
  11. peopleDF <- read.json(sqlContext, path)
  12. printSchema(peopleDF)
  13. # Register this DataFrame as a table.
  14. registerTempTable(peopleDF, "people")
  15. # SQL statements can be run by using the sql methods provided by sqlContext
  16. teenagers <- sql(sqlContext, "SELECT name FROM people WHERE age >= 13 AND age <= 19")
  17. # Call collect to get a local data.frame
  18. teenagersLocalDF <- collect(teenagers)
  19. # Print the teenagers in our dataset
  20. print(teenagersLocalDF)
  21. # Stop the SparkContext now
  22. sparkR.stop()

sparkR  --master yarn-client dataframe.R  这样就可以将任务提交到yarn上了。

另外，有些集群会报如下错误：

  1. 16/06/16 11:40:35 ERROR RBackendHandler: json on 15 failed
  2. Error in invokeJava(isStatic = FALSE, objId$id, methodName, ...) :
  3. java.lang.RuntimeException: Error in configuring object
  4. at org.apache.hadoop.util.ReflectionUtils.setJobConf(ReflectionUtils.java:109)
  5. at org.apache.hadoop.util.ReflectionUtils.setConf(ReflectionUtils.java:75)
  6. at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:133)
  7. at org.apache.spark.rdd.HadoopRDD.getInputFormat(HadoopRDD.scala:185)
  8. at org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:198)
  9. at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)
  10. at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)
  11. at scala.Option.getOrElse(Option.scala:120)
  12. at org.apache.spark.rdd.RDD.partitions(RDD.scala:237)
  13. at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35)
  14. at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)
  15. at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)
  16. at scala.Option.getOrElse(Option.scala:120)
  17. at org.apache.spark.rdd.RDD.partitions(
  18. Calls: read.json -> callJMethod -> invokeJava

这种情况一般是由于启用了lzo压缩导致的。可以通过--jars  添加lzo的jar包，就可以了。例如：sparkR  --master yarn-client dataframe.R --jars /usr/local/Hadoop/share/hadoop/common/hadoop-lzo-0.4.20-SNAPSHOT.jar