Spark中如何生成Avro文件

2022-12-15 07:16:52

研究spark的目的之一就是要取代MR，目前我司MR的一个典型应用场景即为生成Avro文件，然后加载到HIVE表里，所以如何在Spark中生成Avro文件，就是必然之路了。

我本人由于对java不熟，对hadoop的各个框架也知之寥寥，所以耗费了五六天的时间才搞明白怎么在spark中生成avro文件。其中困难有二，一是网上关于spark中生成avro的资料太少，无论官网还是各种论坛，要么没有讯息，要么就是三言两语，对于我这个菜鸟而言，真是要了老命；二是在spark生成avro的代码中，用到了avro框架和hadoop框架的东西，他们自己底层的引用，又有可能和spark的底层引用冲突，虽然最终解决了问题，但是对于问题的直接原因，还么有弄明白。

对于Java的老手，对于hadoop的生态又比较熟悉的人，估计这个课题一天之内就解决了。这里我不怕大家笑话，将目前在本地能跑成功的代码贴出来，还多请指教。还没有提交到集群中去。

1.代码片段

2.pom文件

3.avro格式和文本

1.代码片段

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

import org.apache.spark.SparkContext._

import org.apache.spark.rdd._

import org.apache.hadoop.mapreduce.Job

import org.apache.hadoop.io.NullWritable

import org.apache.avro.Schema

import org.apache.avro.mapred.AvroKey

import org.apache.avro.mapreduce.AvroKeyOutputFormat

import org.apache.avro.mapreduce._

object TextTest extends App {

  System.setProperty("hadoop.home.dir", "D:\\bd\\software\\winutils")

  val sparkConf = new SparkConf().setAppName("AvroTest").setMaster("local[4]")

  val sc = new SparkContext(sparkConf)

  //**************************to generate an avro file based on internal java type

  var li = List("A","A","C","B")

  var lip = sc.parallelize(li, 4)

  var liprdd = lip.map { x => (new AvroKey[String](x),NullWritable.get()) }

  var prdd = new PairRDDFunctions(liprdd)

  val schema = Schema.create(Schema.Type.STRING)

  val job1 = Job.getInstance

  AvroJob.setOutputKeySchema(job1, schema)

  prdd.saveAsNewAPIHadoopFile("D:/002", classOf[AvroKey[String]], classOf[NullWritable],

      classOf[AvroKeyOutputFormat[String]], job1.getConfiguration)

  println("job1 done")

  //**************************to generate an avro file based on avro type

  var av = sc.textFile("D://bdp//NewHuman//Users.txt",5)

  var job = Job.getInstance

  AvroJob.setOutputKeySchema(job, User.getClassSchema)

  val avArray = av.map(x => x.split(" "))

  val userP = avArray.map { x => (new AvroKey[User](User.newBuilder().setFavoriteNumber(Integer.parseInt(x(2))).setName(x(0)).setFavoriteColor(x(1)).build()),NullWritable.get()) }

  var avP = new PairRDDFunctions(userP)

   avP.saveAsNewAPIHadoopFile("D:/003", classOf[AvroKey[User]], classOf[NullWritable],

      classOf[AvroKeyOutputFormat[User]], job.getConfiguration)

  println("job2 done")

}

　　代码中演示了两种类型的场景，一种是内存技术的，一种是外部文件。其中需要注意的是，必须要用AvroJob去设定schema，再者就是只有pairRDD才有saveAsNewAPIHadoop方法，所以其他的RDD必须要转成PairRDD。

　　另外，上面代码中的User类是利用avro自动生成的，需要引用进来。

2.pom文件

 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

     <modelVersion>4.0.0</modelVersion>

     <dependencies>

         <dependency>

             <groupId>jdk.tools</groupId>

             <artifactId>jdk.tools</artifactId>

             <version>1.7.0_67</version>

             <scope>system</scope>

             <systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>

         </dependency>

         <dependency>

         <groupId>com.google.guava</groupId>

         <artifactId>guava</artifactId>

         <version>16.0.1</version>

     </dependency>

         <dependency>

             <groupId>org.apache.spark</groupId>

             <artifactId>spark-core_2.10</artifactId>

             <version>1.2.2</version>

         </dependency>

         <dependency>

             <groupId>org.scala-lang</groupId>

             <artifactId>scala-library</artifactId>

             <version>2.10.4</version>

         </dependency>

         <dependency>

             <groupId>junit</groupId>

             <artifactId>junit</artifactId>

             <version>3.8.1</version>

             <scope>test</scope>

         </dependency>

         <dependency>

             <groupId>org.apache.kafka</groupId>

             <artifactId>kafka_2.10</artifactId>

             <version>0.8.1.1</version>

         </dependency>

         <dependency>

             <groupId>org.apache.spark</groupId>

             <artifactId>spark-streaming_2.10</artifactId>

             <version>1.2.2</version>

         </dependency>

         <dependency>

             <groupId>org.apache.spark</groupId>

             <artifactId>spark-streaming-kafka_2.10</artifactId>

             <version>1.2.2</version>

         </dependency>

         <dependency>

             <groupId>org.apache.avro</groupId>

             <artifactId>avro</artifactId>

             <version>1.7.4</version>

         </dependency>

         <dependency>

             <groupId>org.apache.avro</groupId>

             <artifactId>avro-mapred</artifactId>

             <version>1.7.4</version>

             <classifier>hadoop2</classifier>

         </dependency>

         <dependency>

             <groupId>org.apache.hadoop</groupId>

             <artifactId>hadoop-common</artifactId>

             <version>2.2.0</version>

         </dependency>

         <dependency>

             <groupId>org.apache.hadoop</groupId>

             <artifactId>hadoop-core</artifactId>

             <version>1.0.4</version>

         </dependency>

     </dependencies>

     <repositories>

         <repository>

             <id>scala-tools.org</id>

             <name>Scala-tools Maven2 Repository</name>

             <url>http://scala-tools.org/repo-releases</url>

         </repository>

     </repositories>

     <pluginRepositories>

         <pluginRepository>

             <id>scala-tools.org</id>

             <name>Scala-tools Maven2 Repository</name>

             <url>http://scala-tools.org/repo-releases</url>

         </pluginRepository>

     </pluginRepositories>

     <build>

         <sourceDirectory>src</sourceDirectory>

         <pluginManagement>

             <plugins>

                 <plugin>

                     <groupId>net.alchim31.maven</groupId>

                     <artifactId>maven-scala-plugin</artifactId>

                     <version>3.2.0</version>

                 </plugin>

                 <plugin>

                     <groupId>org.apache.maven.plugins</groupId>

                     <artifactId>maven-compiler-plugin</artifactId>

                     <version>3.1</version>

                 </plugin>

             </plugins>

         </pluginManagement>

         <plugins>

             <plugin>

                 <groupId>org.scala-tools</groupId>

                 <artifactId>maven-scala-plugin</artifactId>

                 <executions>

                     <execution>

                         <id>scala-compile-first</id>

                         <phase>process-resources</phase>

                         <goals>

                             <goal>add-source</goal>

                             <goal>compile</goal>

                         </goals>

                     </execution>

                 </executions>

             </plugin>

         </plugins>

     </build>

 </project>

这个文件中要注意的是dependency的顺序不能变，由于spark和avro的底层的外部引用可能会冲突。

3.avro格式和文本

avro格式为

{"namespace": "example.avro",

 "type": "record",

 "name": "User",

 "fields": [

     {"name": "name", "type": "string"},

     {"name": "favorite_number",  "type": ["int", "null"]},

     {"name": "favorite_color", "type": ["string", "null"]}

 ]

}

测试用例的文本Users.txt

Josh Green 13
Ken Yellow 6
Xiao Orange 8
Gerry Black 12

上一篇：objective-c基础教程——学习小结


下一篇：Java7和8在虚拟机上的差异：Perm Generation vs. Metaspace