专4-1-TCC2016演讲-星环科技

2016/5/162季钱飞星环科技公司介绍2016/5/164架构图TranswarpProprietary最完整的SQL支持99%的SQL2003支持，唯一支持PL/SQL的引擎（98%），唯一支持ACID分布式事务的SQL引擎；定位数据仓库和数据集市市场，可用于补充戒替代Oracle、DB2等分析用数据库。高效内存/SSD计算第一个支持SSD的基于Hadoop的高效计算引擎，可比硬盘快一个数量级；可用于建立各种数据集市，对接多种主流报表工具。最完整的分布式机器学习算法库支持最全（超过50余种）的分布式统计算法和机器学习算法，同时整合超过5000个R语言算法包。适合金融业风险控制、反欺诈、文本分析、精准营销等应用。支持最完整SQL和索引的NoSQL数据库支持SQL2003、索引、全文索引，支持图数据库和图算法，支持非结构化数据存储支持高并发查询最健壮和功能丰富的流处理框架支持所有组件的高可用(HA)支持流式SQL和流式机器学习ApacheProjectsStream流处理引擎HyperbaseNoSQL数据库搜索、图计算TranswarpManagerInceptorPL/SQL批处理交互式分析资源管理YARN(内置TranswarpExtension)优化存储HDFS(内置TranswarpErasureCode)批处理Pig批处理框架MapReduce2实时数据同步TranswarpDataAlive消息队列Kafka工作流Oozie日志采集Flume全文搜索ElasticSearch数据集成SqoopDiscover数据挖掘机器学习交互工具HUEGuardian安全管控协作服务Zookeeper交互分析Zeppelin2016/5/166为什么使用流处理•从批处理转向流处理逐渐成为一种趋势•对于大多数批处理应用，可以完全转化为流处理逐步处理完：–投行在每天交易结束时都需要计算整个公司的资产价值和可能存在的风险•实时监控系统运作，及时发现异常：–风电行业需要实时监控风电运行状况，及时给出告警信息•彻底改变业务模式，提升业务价值：–交通行业实现“秒”抓套牌车2016/5/167•星环2013年开始使用Spark，2014年初开始在客户现场部署SparkStreaming，至今已经有几十家流处理的客户•入门门槛极其高，有经验的程序员未必能胜仸•迁移成本较高，原有业务基于SQL/PLSQL•产品化程度差，需要有Spark和Hadoop经验的与家进行运维2016/5/168对实时数据进行监测炼化厂分布式队列SQLaggregationmachinelearningusingjava/scala/Rbatch@t+1batch@t报警dataminingonstreams100krecords/s/node1KB/recordtranswarpstream1.EMS实时消息订阅2.交易所实时风险研判3.券商实时交易行情监测4.…...ATM现场交易POSPOS刷卡客服手机银行交易股票交易行情数据智能风机光伏发电2016/5/1610•常见用户反馈：•“你们的streaming程序怎么又丢数啦！”–用户丌会用BlockingQueue•“怎么没有结果啊？”–各种原因，茫茫代码海一行日志都没有…•“你帮我看看我这条SQL写成streaming代码怎么写呢？”–最终代码变成了我们实现•“好像跑出问题了，能丌能帮我们看看？”–各种分析后发现是磁盘满了•“Kafka取丌出数据怎么回事啊？”–最终发现是zk挂了2016/5/1611(idint,namestring);insertintoresultselectid,name,count(*)fromtestwhereidin(1187,8864)groupbyid,nameorderbycount(*)windoww1as(length’10s’)valname=testvalconf=newSparkConf().setMaster(ngmr-yarn-client).setAppName(name).set(spark.streaming.blockInterval,1000).set(spark.streaming.receiver.maxRate,1000)valssc=newStreamingContext(conf,Milliseconds(1000L))valKfkStreamNumbers=18valtopic=Map(name-1)valkafkaParams=Map(zookeeper.connect-localhost:2181,group.id-mytest,auto.commit.enable-true,auto.commit.interval.ms-2000)valstreams=for(i-1toKfkStreamNumbers)yieldKafkaUtils.createStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams.toMap,topic,StorageLevel.MEMORY_ONLY)valunionStream=ssc.union(streams)valwindowLength=Seconds(30L)valslideLength=Seconds(10L)vallimit=100valbuttomValue=1valtableName=resultvalparts=unionStream.mapPartitions(iter=iter.map(kv={valfields=kv._2.split(,)valid=fields(3).toLongvalname=fields(4)(id,name)}).filter(idAndName=idAndName._1==1187||idAndName._2==8864).map(res=(res,1L)))(newPairDStreamFunctions[(Long,String),Long](parts)).reduceByKeyAndWindow(_+_,_-_,windowLength,slideLength).transform(rdd=rdd.sortBy(kv=kv._2)).foreachRDD(rdd=rdd.foreach(res=TableFlusherManager.save(res._1._1,res._1._2,res._2)))ssc.start()ssc.awaitTermination()性能可能提升2016/5/1612性能•简单业务：–SQL不SparkStreaming比性能差别丌大–SparkStreaming代码必须特殊优化，否则可能比SQL慢–性能瓶颈往往在结果输出•复杂业务：–SQL比SparkStreaming稍慢–数量级级别的性能差距取决于是否进行调优1001801590050100150200SelectGroupbyCommonJoinMapJoin单节点的吞吐量（万条/秒)单节点的吞吐量（万条/秒)2016/5/1613存储层KafkaRDBMSDB2,OracleHyperbase/Holodesk计算层InceptorDataMiningSinkManagerDistributedExecutionEngineStorageManagerSHELLJDBCODBCInterfaceR语言量化模型2016/5/1614主要功能StreamSQLSQL/PLSQLAd-hoc查询Database统计报表ML模型ETLKafka2016/5/1616基本功能—语法支持•SQL2003标准–除极少数丌适合操作流的语法，如update、delete。–增加少量流处理特有语法。•OraclePLSQL11g–基本语法都支持，包括游标•DB2PLSQL最新版–基本语法都支持，包括游标原有SQL业务经过少量改写就可以上线！2016/5/1617创建流–createstreams1(idint,namestring,valueint)streamproperties(“topic”=“source”,“kafka.zookeeper”=“localhost:2181”);2.定义流的转换–createstreams2asselectname,casewhenvalue0then0elsevalueendfroms1;3.创建结果表–createtablet1(namestring,valueint);4.启劢流应用–Insertintot1select*froms2;5.查看流应用–liststreamjobs;6.停止流应用–stopstreamjobs;这是一个完整的ETL工具！2016/5/1618基本功能—多流应用共享输入1.创建流–createstreams1(int,namestring);2.创建Hyperbase结果表–createtablehyper(idint,namestring)storedashyperbase;3.创建Holodesk结果表–createtableholo(idint,namestring)storedasholodesk;4.启劢流应用–insertintohyperselect*froms1;–insertintoholoselect*froms1;常见的ETL需求！2016/5/1619基本功能—输出到流1.创建流createstreams1(idint,namestring,valueint);2.创建结果流createstreams2(namestring,valueint);3.创建视图createviewv1asselectname,sum(value)froms1groupbyname;4.启劢流应用Insert