AI导航网
微信公众号搜"智元新知"关注
微信扫一扫可直接关注哦！

20.Avro与Spark

分类：Spark作者：编程之家原文地址

20.1 Apache Arvo简介与实验介绍

  • Apache Avro 是一个数据序列化系统，Avro提供Java、Python、C、C++、C#等语言API接口，下面通过java的一个实例来说明Avro序列化和反序列化数据。
    • 支持丰富的数据结构
    • 快速可压缩的二进制数据格式
    • 存储持久数据的文件容器
    • 远程过程调用（RPC）
    • 动态语言的简单集成
  • 实验介绍
    • 如何使用java生成Avro格式数据以及如何通过spark将Avro数据文件转换成DataSet和DataFrame进行操作。

20.2 数据生成—Avro

  • 定义Schema文件
    • 下载avro-tools-1.8.1.jar
Avro官网：http://avro.apache.org/
 Avro版本：1.8.1
 下载Avro相关jar包：avro-tools-1.8.1.jar 该jar包主要用户将定义好的schema文件生成对应的java文件

{
   "namespace":"com.peach.arvo",
  "type": "record",
  "name": "CustomerAddress",
  "fields": [
     {"name":"ca_address_sk","type":"long"},
     {"name":"ca_address_id","type":"string"},
     {"name":"ca_street_number","type":"string"},
     {"name":"ca_street_name","type":"string"},
    {"name":"ca_street_type","type":"string"},
     {"name":"ca_suite_number","type":"string"},
     {"name":"ca_city","type":"string"},
     {"name":"ca_county","type":"string"},
     {"name":"ca_state","type":"string"},
     {"name":"ca_zip","type":"string"},
    {"name":"ca_country","type":"string"},
     {"name":"ca_gmt_offset","type":"double"},
     {"name":"ca_location_type","type":"string"}
  ]
}
  • Schema说明：
    • namespace：在生成java文件时import包路径
    • type：omplex types（record, enum,array, map, union, and fixed）
    • name：生成java文件时的类名
    • fileds：schema中定义的字段及类型
  • 生成java代码文件
    • 使用第1步下载的avro-tools-1.8.1.jar包，生成java code
java -jar  avro-tools-1.8.1.jar compile schema CustomerAddress.avsc .

<dependency>
     <groupId>org.apache.avro</groupId> 
     <artifactId>avro</artifactId> 
     <version>1.8.1</version>   
</dependency>

package com.peach; 
import java.io.BufferedReader; 
import java.io.File; 
import java.io.FileInputStream; 
import java.io.InputStreamReader; 
import java.util.StringTokenizer; 
import  org.apache.avro.file.DataFileWriter; 
import  org.apache.avro.io.DatumWriter; 
import  org.apache.avro.specific.SpecificDatumWriter;   
import com.peach.arvo.CustomerAddress;  

/**   
 *  @author  peach 
 *  2017-03-02 
 * 主要用于生成avro数据文件 
 */ 
public class GenerateDataApp { 
//     private static String customerAddress_avsc_path; 
// 
//     static { 
//         customerAddress_avsc_path =    GenerateDataApp.class.getClass().getResource("/CustomerAddress.avsc").getPath();
//     } 
     private static String source_data_path =  "F:\\data\\customer_address.dat"; //源数据文件路  径 
     private static String dest_avro_data_path =  "F:\\data\\customeraddress.avro"; //生成的avro数据文件路径 

    public  static void main(String[] args) { 

         try { 
//            if(customerAddress_avsc_path !=  null) { 
//                File file = new  File(customerAddress_avsc_path); 
//                Schema schema = new  Schema.Parser().parse(file); 
//            }   
             DatumWriter<CustomerAddress> caDatumwriter = new  SpecificDatumWriter<>(CustomerAddress.class); 
             DataFileWriter<CustomerAddress> dataFileWriter = new  DataFileWriter<>(caDatumwriter);   
             dataFileWriter.create(new CustomerAddress().getSchema(), new  File(dest_avro_data_path)); 
             loadData(dataFileWriter); 
             dataFileWriter.close(); 
         } catch (Exception e) { 
             e.printstacktrace(); 
         } 
     } 

     /** 
      * 加载源数据文件 
      * @param dataFileWriter 
      */ 
     private static void loadData(DataFileWriter<CustomerAddress>  dataFileWriter) { 
         File file = new File(source_data_path); 
         if(!file.isFile()) { 
             return; 
         }  
         try { 
             InputStreamReader isr = new InputStreamReader(new  FileInputStream(file)); 
             BufferedReader reader = new BufferedReader(isr); 
             String line; 
             CustomerAddress address; 
             while ((line = reader.readLine()) != null) { 
                address =  getCustomerAddress(line); 
                if (address != null) { 
                     dataFileWriter.append(address);   
                } 
             } 
             isr.close(); 
             reader.close(); 
         } catch (Exception e) { 
             e.printstacktrace(); 
         } 
     } 

     /** 
      * 解析单条文本数据封装CustomerAddress对象 
      * @param line 
      * @return 
      */ 
     private static CustomerAddress getCustomerAddress(String line) { 
         CustomerAddress ca = null; 
         try { 
             if (line != null && line != "") { 
                StringTokenizer token = new  StringTokenizer(line, "|"); //使用stringtokenizer拆分字符串时，会去自动除""类型 
                 if(token.countTokens()  >= 13) { 
                    ca = new  CustomerAddress(); 
                     ca.setCaAddressSk(Long.parseLong(token.nextToken())); 
                     ca.setCaAddressId(token.nextToken());   
                    ca.setCaStreetNumber(token.nextToken()); 
                     ca.setCaStreetName(token.nextToken());   
                     ca.setCaStreetType(token.nextToken());   
                     ca.setCaSuiteNumber(token.nextToken()); 
                    ca.setCaCity(token.nextToken()); 
                     ca.setCaCounty(token.nextToken());   
                     ca.setCaState(token.nextToken());   
                     ca.setCaZip(token.nextToken());   
                     ca.setCaCountry(token.nextToken());   
                     ca.setCaGmtOffset(Double.parseDouble(token.nextToken())); 
                     ca.setCaLocationType(token.nextToken()); 
                } else { 
                     System.err.println(line); 
                } 
             } 
        } catch (NumberFormatException e)  { 
             System.err.println(line); 
         } 

         return ca; 
     } 
}
  • 动态生成avro文件，通过将数据封装为GenericRecord对象，动态的写入avro文件，以下代码片段：
private static void  loadData(DataFileWriter<GenericRecord> dataFileWriter, Schema schema)  { 
     File file = new File(sourcePath);   
     if(file == null) { 
         logger.error("[peach], source data not found"); 
         return ; 
     } 

     InputStreamReader inputStreamReader = null; 
     BufferedReader bufferedReader = null;   
     try { 
         inputStreamReader = new InputStreamReader(new  FileInputStream(file)); 
         bufferedReader = new BufferedReader(inputStreamReader); 
         String line; 
         GenericRecord genericRecord; 
         while((line = bufferedReader.readLine()) != null) { 
             if(line != "") { 
                String[] values =  line.split("\\|"); 
                genericRecord =  SchemaUtil.convertRecord(values, schema);   
                if(genericRecord != null) { 
                     dataFileWriter.append(genericRecord);   
                } 
             } 
         } 

     } catch (Exception e) { 
         e.printstacktrace(); 
     } finally { 
         try { 
             if(bufferedReader != null) { 
                bufferedReader.close(); 
             } 
             if(inputStreamReader != null) {   
                 inputStreamReader.close(); 
             } 
         } catch (IOException e) { 
         } 
     } 
}

20.3 读Avro文件—Spark

<dependency> 
     <groupId>com.peach</groupId> 
     <artifactId>generatedata</artifactId> 
     <version>1.0-SNAPSHOT</version> 
</dependency> 
<dependency> 
     <groupId>com.databricks</groupId> 
    <artifactId>spark-avro_2.10</artifactId> 
    <version>2.1.0</version> 
</dependency> 
<!--  https://mvnrepository.com/artifact/org.apache.spark/spark-sql_2.10 -->
<dependency> 
     <groupId>org.apache.spark</groupId> 
     <artifactId>spark-sql_2.10</artifactId> 
    <version>2.1.0</version> 
</dependency> 
<dependency> 
     <groupId>org.apache.spark</groupId> 
     <artifactId>spark-core_2.10</artifactId> 
    <version>2.1.0</version> 
</dependency> 
<dependency> 
     <groupId>org.apache.avro</groupId> 
    <artifactId>avro</artifactId> 
    <version>1.8.1</version> 
</dependency>

case class  CustomerAddressData(ca_address_sk: Long,
                               ca_address_id:  String,
                                ca_street_number: String,
                                ca_street_name: String,
                                ca_street_type: String,
                                ca_suite_number: String,
                               ca_city:  String,
                               ca_county: String,
                               ca_state:  String,
                               ca_zip:  String,
                               ca_country:  String,
                               ca_gmt_offset:  Double,
                                ca_location_type: String
                              ) 
//   org.apache.spark.sql.catalyst.encoders.OuterScopes.addOuterScope(this) 

def main(args:  Array[String]): Unit = { 
  val path =  "/Users/zoulihan/Desktop/customeraddress.avro" 
  val conf = new SparkConf().setAppName("test").setMaster("local[2]") 
  val sc = new SparkContext(conf) 
  val sqlContext = new sqlContext(sc) 
  import sqlContext.implicits._ //为什么要加此段代码？ 

  val _rdd =  sc.hadoopFile[AvroWrapper[CustomerAddress], NullWritable, AvroInputFormat[CustomerAddress]](path) 
  val ddd = _rdd.map(line => new  CustomerAddressData( 
    line._1.datum().getCaAddressSk, 
     line._1.datum().getCaAddressId.toString, 
     line._1.datum().getCaStreetNumber.toString, 
    line._1.datum().getCaStreetName.toString, 
     line._1.datum().getCaStreetType.toString, 
     line._1.datum().getCaSuiteNumber.toString, 
    line._1.datum().getCaCity.toString, 
     line._1.datum().getCaCounty.toString,   
    line._1.datum().getCaState.toString, 
    line._1.datum().getCaZip.toString, 
     line._1.datum().getCaCountry.toString,   
    line._1.datum().getCaGmtOffset, 
     line._1.datum().getCaLocationType.toString 
  )) 
  val ds = sqlContext.createDataset(ddd) 
  ds.show()   
  val df = ds.toDF(); 
  df.createTempView("customer_address");
//    sqlContext.sql("select count(*) from  customer_address").show()
  sqlContext.sql("select * from  customer_address limit 10").show()
}

大数据视频推荐：
CSDN
大数据语音推荐：
企业级大数据技术应用
大数据机器学习案例之推荐系统
自然语言处理
大数据基础
人工智能：深度学习入门到精通

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。

相关推荐

1_Spark Streaming 概述
1.SparkStreaming是什么？SparkStreaming是SparkCore的扩展API用来支持高吞吐、高容错的处理流式数据数据源可以是:Kafka、TCPsockets、Flume、Twitter等流式数据源处理数据:可以用SparkCore的算子map、reduce、join、...
Spark通讯录相似度计算怎么实现
本篇内容介绍了“Spark通讯录相似度计算怎么实现”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这...
如何进行Spark数据分析
本篇文章给大家分享的是有关如何进行Spark数据分析，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说...
Spark Shuffle和Hadoop Shuffle有哪些区别
本篇内容主要讲解“Spark Shuffle和Hadoop Shuffle有哪些区别”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“S...
TSDB的数据怎么利用Hadoop/spark集群做数据分析
这篇文章主要介绍“TSDB的数据怎么利用Hadoop/spark集群做数据分析”，在日常操作中，相信很多人在TSDB的数据怎么利用Hadoop/spark集群做数据分析问题上存在疑惑...
Hadoop与Spark性能原理是什么
本篇内容介绍了“Hadoop与Spark性能原理是什么”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这
Hadoop和Spark有什么不同
小编给大家分享一下Hadoop和Spark有什么不同，相信大部分人都还不怎么了解，因此分享这篇文章给大家参考一下，希望大家阅读完这篇文章后大有收获，下面让我们
Hadoop和Spark的Shuffle过程有什么不同
这篇文章主要讲解了“Hadoop和Spark的Shuffle过程有什么不同”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习...
基于CDP7.1.1的Spark3.0技术预览版本分析是怎样的
本篇文章给大家分享的是有关基于CDP7.1.1的Spark3.0技术预览版本分析是怎样的，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获...
Spark中foreachRDD、foreachPartition和foreach的区别是什么
这篇文章主要介绍“Spark中foreachRDD、foreachPartition和foreach的区别是什么”，在日常操作中，相信很多人在Spark中foreachRDD、foreachPartition和foreach的...
    pythonJavaScriptjavaHTMLreactjsC#AndroidCSSNode.jssqlrpython-3.xMysqLjQueryc++pandasFlutterangularIOSdjangolinuxswifttypescript路由器JSON路由器设置无线路由器h3c华三华三路由器设置华三路由器电脑软件教程arraysdocker软件图文教程Cvue.jslaravelspring-bootreact-native