个性化系统培训讲义

BDTC2014数据技术小象科20141212@北冥乘、个性 分析–网络(Ad需求方平台(DemandS 基于行为数据的同时关注QPS和快速就意味着效果、意味着收入小比例流量上达不到最优不是严重问题例：Near-linepage避免集中读写形成的单点性能瓶颈增删改查的传统需求实际上不存在个性化系般架 ，见后例：计算系 –业务逻 指不在用户主体需求任务中的产品增删改查的传统需求可以避免例：点击率计算需要的特征可以利用引擎本机内存做redis的少量的决策错误或者超时可以接受多Server的上下线管理和调度by轻量级Web服务器／反向服务Nginx+–将个性化系统的主要其他产品选择Zookeeperby解决分布式应用中的一些数据管理问题器加入等迅速通知到每 创

创EPHEMERAL类型节类搜索的决策过Search文档“”D2=“谷歌地图创始 “”3 (term)倒排链{D1,D2,D3,D4}，地图{D1,D2,D3,D4}，之父{D1,D3,D4}，跳槽{D1,D3}，{D1,D2,D3,D4}，…Query可能相当长，需要利用相关性检索信息的过来吧条件往往表示成一个表达式而非 开源的全文检索引擎不支持相关性检索其他产品选择Elasticby用ThriftIDL定义通 为访客打上业务相关的通过访客协同数据进行个性化–用户响应，为线上排序提供指报表、Cube、日志的吞吐量很大日志收集的结构较复杂日志的一致性要求与交易系统相比稍低、、Thrift-RPC、tail、sslo、、Thrift-RPC、syslog其他产品选择Scribe、分布式与计算平分布式需分布式计算需求用户与模型特征缓User由Hadoop按日更新，或由Storm准实时上下文属性URL(App)用半的方式准实时更点击率模型参数、特征KV型需一个KV系其他产品选择半上下文系用Redis系统url特征表供实时Redis不预先加载任何内容，对不存在的url,立刻返回空特征,同时触发相应的页面爬虫和特征提取设置Redis关于开源的机器学习开源机器学习工具开源机器学习包流计算 流计算不适合什么场景？实时反计费或投放控制实时受众定向实时点击反馈by 数据流基本在网络和内存进行“StormonYARN”可以与Hadoopstreamingstreaming小规模批处理系统可以在YARN上与Hadoop共享计算资源可以更方便地利用Spark全栈的处理工具– Spark mender@CrazyJvmBDTC2014Part1:Spark编程模型及内核解析Part2:SparkStreamingPart3:Part4:MLlib算法与架构实Part1:SparkWhatWhatisApacheSparkisafastandgeneralengineforlarge-scaledataprocessing.EaseofUseIntegratedwithApacheApacheSparkSQL/ApacheApacheoneonestacktorulethemSpark Ad-hocBatchHadoopHadoopDataSer&DeSerDiskIOSparkSparkDataone-WhySparkMemorybasedcomputationThreadOptimization(eg:delayClusterClusterarnSparkSparkInputInputInputKeyKeyConcept-AlistofAfunctionforcomputingeachAlistofdependenciesonotherRDDsOptionallaPartitionerforkey-valueRDDsOptionallalistofpreferredlocationstocomputeeachsplitonFaultFaultWhatisProgrammingProgramming!StoragevalDISK_ONLY=newStorageLevel(true,false,false,valDISK_ONLY_2=newStorageLevel(true,false,false,false,2)valMEMORY_ONLY=newStorageLevel(false,true,false,valMEMORY_ONLY_2=newStorageLevel(false,true,false,true,2)valMEMORY_ONLY_SER=newStorageLevel(false,true,false,false)

默认：内存非序列valMEMORY_ONLY_SER_2=newStorageLevel(false,true,false,false,2)valMEMORY_AND_DISK=newStorageLevel(true,true,false,true)valMEMORY_AND_DISK_2=newStorageLevel(true,true,false,true,2)valMEMORY_AND_DISK_SER=newStorageLevel(true,true,false,false)valMEMORY_AND_DISK_SER_2=newStorageLevel(true,true,false,false,2)valOFF_HEAP=newStorageLevel(false,false,true,false)Narrow Wide效率效率容错vallines=errors.persist()valhbase_error=errors.filter(_.contains(“HBase”)).countvalmysql_error=errors.filter(_.contains(“MySQL”)).count//……(一系列对error的操作ransormtio&transformation& -packagepackageSourceCode(canbuildseveralHadoopversions)ClusterClusterslavesspark-env.shexportSPARK_WORKER_CORES=16exportexportSPARK_WORKER_MEMORY=26gexportSPARK_MASTER_PORT=8888spark-spark-spark.masterspark://ip:portspark.local.dir/data/tmp_spark_dir/spark.executor.memory20gpression.codecsubmitsubmit./bin/spark-submit--class<mainmaster<master-url>...#other<application-jar>\2、spark-submit3、默认的参数配置文件（spark-有有朋友认为这个顺序不合理:Demo(集群ClusterDriverRDDRDDPartitionPartitiononeoneaskRDD

stageBuildoperator

splitgraphintostages

cluster

stragglingsubmiteachstageas

retryfailedexecutestoreandservebackbacktoPartitions:protecteddefgetPartitions:Dependencies:protecteddefgetDependencies:Seq[Dependency[_]]=depscompute:defcompute(split:Partition,context:TaskContext):Iterator[T]PreferredLoc(optional):protecteddefgetPreferredLocations(split:Partition):Seq[String]=NilPartitioner(optional):@transientvalpartitioner:Option[Partitioner]=Nonepartitions每个HDFSblockdependencies:无compute:每一个preferredlocationsHDFSblockpartitionerpartitions:与父RDD一致dependencies与父RDD一对一compute:计算父RDD的每个分区并过滤preferredlocations无(与父RDD一致)partition:无partitions:每个reduce任务一个分区dependencies:依赖所有父RDDcompute: preferredlocations:无partition:目目标计算每个分区的函结果将taskset传给底层调度器重新提交shuffle输出丢失的Scheduler一个Stage内的窄依赖进行pipeline操1+1+1+1=4①1+1=2;2+1=3;3+1=4ask外shuffle

结Staeshfle为了容错,会把hul输出写在磁盘或者内存任何一个任务可以运行在任何一允许任务使用那些被缓存但是已经被置换出去askhedule出现shuffle输出lost要报告fetchfailed错误

SortSort-basedshuffleMasterMaster&orker

BroadcastBroadcastBT形式的广播变使用场lookup表mapside: 只读,存于每台worker的cache,不随task使用方valbroadcastVarsc.broadcast(Array(1,2,3))注意：Broadcastedbinaryforthetaskusedtodispatchtaskstoexecutorssince用法:valaccum=sc.accumulator(0)sc.parallelize(Array(1234)).foreach(xaccumx)Part2:SparkTCPsocketsAkkaactor陆续增加中transformation-transformation-sortByKey,join,etctransformation-transformation-window,countByWindow,reduceByWindowcountByValueAndWindow,DStream输何时清理何时清理对于window和stateful操作默认持对于来自网络的数据源,每份数据会在内存中存两对于window和stateful操作必须checkpont通Sagtot指定通过trot定隔间间隔必须是slide interval的倍数容容oker一定可以通过重新数据来恢复,绝对不会有数据丢默认会在两个不同节点加载数据到内存,一个节点fail了,假设正在运行inputeceier的节点fail了,可能会丢失一部分数据(wy?没来得及)DriverDriver会定期将元数据写到指定的Driver失败后可以通过元数据重新启动valcontext=StreamingContext.getOrCreate(checkpointDirectory,functionToCreateContextgetOrCreate,必须确保每次重新编译后清空 选择合适的选择合适的batch没有最好的sisi:要来得及消化流进系统的数据可以从Log4j或者StreamingListenerCMS(暂停时间短,但吞吐率不高,并且会引起内存碎片) WindowPart3:Kafka(+ApacheKafkaispublish-subscribemessagingrethoughtasadistributedcommitlogOpensourceby组组Kafkaisapersistent,distributed,replicatedpub/submessagingsystem.Producerssendmessagestoaclusterofbrokers.Thebrokerspersistthemessagestodisk.Consumersthenrequestarangeofmessagesusing(offset,length)styleReceivemessagesfromProducers(push),delivermessagestoConsumers(pull)Responsibleforpersistingthemessages(指定一段时间)RelativelylightweightLog-basedLog-basedMessagesarepersistedtoappend-onlylogfilesbythebroker.Producersareappendingtotheselogfiles(sequentialwrite),andconsumersarereadingarangeofthesefiles(sequentialreads).opiDataisstoredintopics,andtopicsaresplitintopartitions,whicharereplicated.kafka-topics.sh--zookeepercloudx:2181--create--topiccc--partition3--replication-factor2--configautocreate:Partitionsofatopicarereplicated.Onebrokeristhe"leader"ofapartition.Allwritesandreadsmustgototheleader.Replicasexistforfault-tolerance,notscalability.Whenwriting,messagescanbesynchronouslywrittentoNreplicas(depend