首页 文章 精选 留言 我的

精选列表

搜索[时序预测],共10000篇文章
优秀的个人博客,低调大师

用Python玩转时序数据

时间序列是日常生活中最常见的数据类型之一。股票价格、销售信息、气候数据、能源使用,甚至个人身高体重都是可以用来定期收集的数据样本。几乎每个数据科学家在工作中都会遇到时间序列,能够有效地处理这些数据是数据科学领域之中的一项非常重要的技能。 本文简要介绍了如何从零开始使用Python中的时间序列。这包括对时间序列的简单定义,以及对利用pandas访问伦敦市居民智能电表所获取数据的处理。可以点击此处获取本文中所使用的数据。还提供了一些我认为有用的代码。 让我们从基础开始,时间序列的定义是这样的: 时间序列是按时间的顺序进行索引、排列或者绘制的数据点的集合。最常见的定义是,一个时间序列是在连续的相同间隔的时间点上取得的序列,因此它是一个离散时间数据的序列。 时间序列数据是围绕相对确定的时间戳而组织的。因此,与随机样本相比,可能包含我们将要尝试提取

优秀的个人博客,低调大师

时序数据库 DolphinDB 线程简介

本文基于 DolphinDB server 最新版 2.00.X,从任务管理、存储引擎、流数据、集群管理、高可用几个方面,为大家简单介绍 DolphinDB 在运行中可能使用到的各种线程,及其相关的配置项和函数,以便用户掌握 DolphinDB 的线程运行情况。 1.任务管理 工作线程(Worker)接收客户端请求,将任务分解为多个小任务,根据任务的粒度自己执行或者发送给执行线程 localExecutor 或 remoteExecutor 执行。 Worker 常规交互作业的工作线程。每个节点都存在 Worker 线程,可以分为以下几类。 ZeroWorker, FirstWorker, SecondWorker, ThirdWorker, ForthWorker 客户端提交至节点的作业为0级,由 ZeroWorker 处理。根据作业所涉及到的分区,ZeroWorker 将其分解为多个子任务。其中本地节点上的子任务由 ZeroWorker 与 localExecutor 并行执行;需要由远程节点执行的子任务则降低为1级,并通过 remoteExecutor 发送到对应节点上的 FirstWorker 处理。 以此类推,若某个级别的子任务需要进一步拆解,则拆分出来的由远程节点执行的子任务降低一级,发送至远程节点上对应层级的 Worker 处理。 ZeroWorker 和 FirstWorker 的数量由 workerNum 决定,默认值为机器上的 CPU 核数,最大值不超过 license 中的最大 核数。其余层级的 Work 数量为上级的0.75倍,最小个数为1。 UrgentWorker 处理紧急任务,只接收一些特殊的系统级任务,譬如登录,取消作业等。由 urgentWorkerNum 配置,默认值为1,最大值为 CPU 内核数。 WebWorker 处理 HTTP 请求,由 webWorkerNum 配置。默认为1,最大值为 CPU 内核数。 InfraWorker 开启高可用后,用于接收 raft 心跳汇报的线程,防止集群负载大时,心跳信息无法及时汇报。默认有2个该线程。 LocalExecutor 本地执行线程。Worker 拆解完任务后生成的本地子任务队列,由同一节点下的 localExecutor 执行。所有 Worker共享本地执行线程,每个 localExecutor 一次只能处理一个子任务。通过localExecutors配置线程个数,默认为机器上的 CPU 核数减1。 RemoteExecutor 将远程任务发送到远程节点的线程,在非 single 模式的节点上可以通过remoteExecutors配置线程个数。默认值为集群中节点个数和本地 Worker 的较小值。 AsynchronousRemoteExecutor 接收对其他节点发起的远程调用(Remote Procedure Call, RPC)任务的线程,并将收到的远程调用放到 RemoteExecutor 的任务队列中。每个非 single 模式的节点上有且仅有一个该线程。 RemoteTaskDispatcher 在远程调用出错需要重试时,或者一个被关闭的连接上仍有未完成的任务时,这些任务会先放到一个队列里,由 RemoteTaskDispatcher 从这个队列取任务并重新交由 AsynchronousRemoteExecutor 去发起远程调用。 DynamicWorkerManager 和 DynamicWorker DynamicWorker 是动态工作线程,作为 Worker 的补充。DynamicWorkerManager 是创建 DynamicWorker 的线程,每个节点有且仅有一个该线程。如果所有的工作线程被占满,有新任务到来时,通过该线程创建 DynamicWorker 来执行新任务。根据系统并发任务的繁忙程度,总共可以创建三组动态工作线程,每一个级别可以创建 maxDynamicWorker 个动态工作线程。 动态工作线程在任务执行完后若闲置60秒则会被系统自动回收,不再占用系统资源。maxDynamicWorker 的默认值为 workerNum。 DynamicExecutorManager 和 DynamicExecutor 与 DynamicWorkerManager 和 DynamicWorker 类似,DynamicExecutor 是由 DynamicWorkerManager 动态创建的执行线程。DynamicExecutor 线程数量上限由maxDynamicLocalExecutor决定,默认值为 localExecutors。DynamicWorkerManager 最多可以创建3组动态执行线程,每组线程最多为 maxDynamicLocalExecutor 个。 DynamicExecutor 在闲置60秒后被自动回收。 BlockIOWorker 执行对硬盘读写任务的线程。通过diskIOConcurrencyLevel控制线程数量,默认值为1。 BatchJobWorker 执行批处理作业任务的工作线程。其上限通过maxBatchJobWorker设置,默认值是 workerNum。该线程在任务执行完后若闲置60秒会被系统自动回收,不再占用系统资源。 2.存储引擎 存储相关的线程在 server 启动过程中创建。其主要负责数据的写入与落盘,并在各种情况下(如节点宕机,磁盘损坏导致数据损坏)维护各节点间数据的完整与一致性。 2.1 预写日志 在数据节点上,OLAP 和 TSDB 都有两个写预写日志(WAL)的线程 RedoLogHeadWriter 和 RedoLogDataWriter,分别负责写 WAL 的元数据和数据。 存储 OLAP 和 TSDB WAL 日志的目录分别由redoLogDir,TSDBRedoLogDir配置。 RedoLogHeadWriter 事务发生时通过 RedoLogHeadWriter 同步写事务元数据信息到 redoLog 目录下的 header.log 中。 RedoLogDataWriter 事务的数据会通过 RedoLogDataWriter 异步写入 redoLog 目录下的 tid.log 中。 2.2 OLAP 引擎 在开启了 OLAP 的 cacheEngine 后,会创建一个 ChunkCacheEngineGCWorker 线程。 ChunkCacheEngineGCWorker 负责将 cacheEngine 中的数据写入磁盘的线程。用于清理 cacheEngine,并将磁盘的随机写变成顺序写的线程。BackgroundRoutineService 每隔60秒,或者写入时 cacheEngine 的占用量超过 OLAPCacheEngineSize 的30%,就会触发 ChunkCacheEngineGCWorker 将 cacheEngine 中的数据写入磁盘。cacheEngine 中一张表的一个分区数据称为一个 tabletCache,ChunkCacheEngineGCWorker 写入磁盘时会根据每个 tabletCache 的大小和在 cacheEngine 中存在时间决定写入磁盘优先级,tabletCache 的大小越大、存在时间越久写入磁盘优先级越高。 2.3 TSDB 引擎(1.30.X 版本 server 无此类线程) 除了将数据写入磁盘外,TSDB 引擎的线程还要负责 cacheEngine 中数据的排序与合并,并维护磁盘上的 levelFile,以提高读写性能。这些线程也仅在数据节点上存在。 TableAsyncSortRunner 异步地对 TSDB cacheEngine 中的表进行排序的线程。TSDB 在写入 cacheEngine 时,如果 cacheEngine 中的表太大,会影响查询性能,因此需要进行排序。但若同步进行排序,会影响写入性能。所以 DolphinDB 提供此线程异步地对表进行排序。可以通过TSDBAsyncSortingWorkerNum来控制排序线程的数量,默认值为1。也可以通过函数disableTSDBAsyncSorting和enableTSDBAsyncSorting,来手动开启和关闭异步排序功能。 CacheDumpTaskDispatcher 分配 cacheEngine 写入磁盘任务的线程。线程数量固定为1。 当 cacheEngine 的内存占用大于 TSDBCacheEngineSize 时,系统会对 cacheEngine 做一次快照,并将快照送到 CacheDumpTaskDispatcher 线程准备写入磁盘。CacheDumpTaskDispatcher 线程将任务分配给 ChunkCacheDumpRunner 线程,由该线程写入磁盘。若磁盘上存在需要合并的 levelFile,则交由 MergeRunner 线程进行合并。 ChunkCacheDumpRunner 将 cacheEngine 中的数据写入磁盘的线程。线程的个数等于volumes的配置值。 MergeRunner 对磁盘上 levelFile 进行合并的线程,线程的个数等于volumes的配置值。 DelLevelFileRunner 检查并删除无效的 levelFile(即已经被合并的较小 size 的文件)的线程。线程数量固定为1。每隔30秒会自动执行一次。 2.4 数据恢复 数据恢复(recovery)相关线程负责节点宕机,或者数据损坏时,数据副本间的数据恢复。 RecoveryReportingService 在数据节点上,任何一个 chunk 发生数据错误或者版本号不一致都通过该线程来向控制节点汇报。每个数据节点有且仅有一个该线程。 RecoveryWorker 发生 recovery 时,数据恢复的源节点将数据发送给目标节点的线程。该线程仅存在于数据节点,个数可由recoveryWorkers配置,默认值为1。可以通过resetRecoveryWorkerNum函数动态修改线程个数,通过getRecoveryWorkerNum函数获取实际 RecoveryWorker 线程数量。 RecoverMetaLogWriter 和 RecoverRedoLogDataWriter 在线恢复(onlineRecovery)过程中,为了避免节点宕机或离线影响恢复过程,会分别通过 RecoverMetaLogWriter 和 RecoverRedoLogDataWriter 写 recover redoLog 的元数据(Metadata)和数据(data)。与 redoLog 不同的是,recover redoLog 的 Metadata 和 data 需要进行写磁盘时才能开始 recovery。通过enableDfsRecoverRedo配置是否开启 recover redoLog,默认是开启。开启后,在每个数据节点上存在一个相应的线程。recover redoLog 的文件目录也可以通过recoverLogDir配置,默认在节点根目录下的log/recoverLog中。 DFSChunkRecoveryWorker 在控制节点上处理 recovery 任务的线程。同时进行 recovery 任务的数量默认为集群中数据节点个数的2倍,可由dfsRecoveryConcurrency配置。 2.5 事务相关 如果集群中的某个节点在处理事务的过程中宕机了,那么重启后仅依靠该节点有可能无法确定事务的最终状态,需要在集群中进行事务决议来确定。 UnresolvedTransactionReporter 在数据节点启动时,如果数据节点自己不能判断某些事务的状态,通过该线程来向控制节点汇报并发起事务决议,判断事务最终处于回滚还是完成状态。该线程只有一个,且在所有需要决议的事务决议后结束。 DFSTransactionResolutionWorker 该线程处理由数据节点发起的事务决议、控制节点启动时回放元数据后无法决定状态的事务或运行时超时未更新状态的事务。在控制节点上存在一个该线程。 ChunkNodeLogWriter 数据节点写元数据的线程。元数据默认在各个数据节点根目录下的storage/CHUNK_METADATA中,可以通过配置项chunkMetaDir修改。在数据节点上存在一个该线程。 EditLogBatchWriter 控制节点写元数据的线程。由于对控制节点上元数据的修改比较频繁,所以由该线程统一将写入缓冲区的数据写入磁盘并同步,同时还对写元数据失败的情况进行回滚处理。在控制节点上存在一个该线程。 2.6 其他 SnapshotEngineWorker 为减少开启快照引擎对写入的影响,而将分布式表数据异步写入快照引擎的线程。在数据节点上存在一个该线程。可以通过函数registerSnapshotEngine和unregisterSnapshotEngine对一个分布式表注册和取消注册快照引擎。 DFSChunkRebalanceWorker 节点间平衡数据或者多块磁盘间平衡数据的任务,均交由控制节点上的 DFSChunkRebalanceWorker 线程处理。在控制节点上存在一个该线程。同时发起的数据平衡任务数量默认为集群中数据节点个数的两倍。可由dfsRebalanceConcurrency配置。通过函数rebalanceChunksWithinDataNode和rebalanceChunksAmongDataNodes手动触发节点或磁盘间的数据平衡。 3.流数据 本节通过发布订阅、计算引擎和高可用三个模块介绍流数据相关线程。这些线程都仅在数据节点或单节点上存在。 3.1 发布订阅 以下为数据节点上普通流表的订阅发布流程中涉及到的线程。 MessageThrottle 实现流数据订阅 throttle 参数功能的线程,数量为1。系统每隔一段时间检查当前节点上是否存在经过 throttle 时间但仍未达到 batchSize 的订阅(subscribeTable函数中指定了 batchSize 和 throttle )。如果存在,则触发一次订阅的消息处理。通过subThrottle配置触发检查的间隔时间,默认值为1000,单位为毫秒。 AsynchronousPublisher 在 AsynchronousPublisher 线程中检查每个发布节点对每个订阅节点建立的连接。如果这个连接对应的发布队列有更新,就将更新的数据发布到订阅端。通过maxPubConnections配置发布节点连接的订阅节点数量上限,默认值为0,表示不可以作为发布节点,即不会创建 AsynchronousPublisher 线程,大于0时会创建一个该线程。 AsynchronousSubscriber 监听所有的订阅连接,接收、解析连接上收到的数据,并发送到相应的订阅消息队列。配置了subPort之后,会创建一个该线程。 通过maxSubConnections配置一个订阅节点可以连接的发布节点数量上限,默认值为64。 LocalSubscriberImp 在 LocalSubscriberImp 线程中直接检查有数据更新的本地订阅,并将符合条件的本地订阅中的数据发送到订阅消息队列中。配置 subPort 后,会创建一个该线程。 StreamExecutor StreamExecutor 线程从订阅消息队列中取出数据,写入相应订阅的 handler 中,同时维护订阅的偏移量、消息总数等信息。每个订阅消息队列对应一个 StreamExecutor 线程,数量由配置项 subExecutors 决定,默认值为1,最大不超过 CPU 核数。 PersistenceWorker 以异步方式持久化的流表会通过 PersistenceWorker 线程将数据写到磁盘上。persistenceWorkerNum控制持久化线程的数量,默认为1。由persistenceDir配置开启持久化的流表的保存路径。 AsynchronousReconnector 针对所有设置参数 reconnect=true 的订阅,系统会在非正常中断后通过该线程尝试自动重连。在配置了subPort之后,会创建一个该线程。 3.2 计算引擎 创建计算引擎时,若配置了如下参数,便会创建两个线程:CheckTimeExecutor 和 SystemTimeExecutor。 CheckTimeExecutor 包括 TimeSeriesCheckTimeExecutor, SessionCheckTimeExecutor, CSEngineCheckTimeExecutor, AsofJoinCheckTimeExecutor 和 LookupJoinCheckTimeExecutor。 在创建 TimeSeriesEngine 时设置了 updateTime、创建 SessionWindowEngine 时设置了 forceTriggerTime、创建 CrossSectionalEngine 时设置了 triggeringPattern=“interval”、创建 AsofJoinEngine 时设置了 delayedTime、创建 LookupJoinEngine 时设置了 checkTimes,那么每个引擎就会创建一个 CheckTimeExecutor 线程,表示如果经过了参数设置的时间还未触发计算,则强制触发一次引擎的计算。 SystemTimeExecutor 包括 TimeSeriesSystemTimeExecutor, SessionSystemTimeExecutor, CrossSectionalEngineExecutor 和 WindowJoinSystemTimeExecutor。 在创建 TimeSeriesEngine, SessionWindowEngine, CrossSectionalEngine 和 WindowJoinEngine 时,如果设置了 useSystemTime=true,那么每个引擎就会创建一个 SystemTimeExecutor 线程,表示每隔固定的时间触发一次引擎的计算。 3.3 流数据高可用 配置项streamingRaftGroups中每个 group 都会在 group 内的节点上生成下述的三个线程。 StreamingDataFileWriter 在 raft 的 leader 节点上向流表写数据时,要通过该线程应用 leader 上写数据的 entryLog,向流表写数据。 StreamingRaftReplayWorker 当一个节点成为某个 group 的 leader 时,就会通过该线程回放此 group 的 raftLog。 StreamingHA::CkptWorker 为节点上的 raftLog 做 checkpoint 以回收垃圾的线程。垃圾回收的间隔可由streamingHAPurgeInterval设置,默认值为300,单位是秒。 4.集群管理 在集群中控制节点通过心跳监控其他节点的存活状态。 HeartBeatSender 控制数据节点或计算节点向控制节点每隔0.5秒发送一次心跳的线程。心跳信息中同时还会汇报节点当前的一些信息(如 CPU、内存、磁盘占用)给控制节点。在数据节点或计算节点上存在一个该线程。 通过 lanCluster 控制心跳采用 udp 或 tcp 协议,当为 true 时使用 udp,false 时使用 tcp,默认值为true。 HeartBeatReceiver 仅当 lanCluster=true 时,在控制节点和数据节点、计算节点上存在的接收 udp 心跳的线程。 HeartBeatMonitor 仅在控制节点存在的线程。每隔一秒检查一次是否收到集群中数据节点或计算节点的心跳信息。如果一个节点连续3次检查都没有心跳,就认为这个节点已经宕机了。 如果数据节点配置了 datanodeRestartInterval(值大于0),那么当节点宕机时间超过设置值,就会通过 agent 重启该数据节点。该配置项默认值为0。 ServiceMgmtCenter 仅在控制节点存在的线程。当一个代理节点重新上线时,通过该线程将公钥信息保存到代理节点上。当一个数据节点重新上线时,会让其汇报节点的所有 chunk 信息,并且在数据节点上删除控制节点上不存在的chunk。 5.控制节点高可用 本节简述开启控制节点高可用之后,raft 相关的线程。对于每种线程,在 raft group 内的每个控制节点上都有且仅有一个。 RaftTimer 负责计时(心跳发送间隔和发起选举时间)的线程。leader 通过该线程每隔一段时间向 follower 发送心跳信息,follower 如果一段时间没有收到 leader 的心跳,将发起选举。 通过raftElectionTick可以设置在 [raftElectionTick, 2*raftElectionTick] 之间的一个随机时间后未收到 leader 的心跳将发起选举,默认值为800,单位是10ms。 RaftInputWorker 从输入消息队列取出消息应用到当前节点的线程。 RaftOutputWorker 从输出消息队列取出消息并应用到相应节点的线程。 RaftProposeWorker 处理对 raftLog 读写请求的线程。 SnapshotSender 将 leader 当前状态的快照发送给其他节点的线程。 RaftLeaderSwitchWorker 执行 raft 节点角色切换的线程。 DFSRaftReplayWorker 将记录的 raftLog 应用到当前节点的线程。 6.其他 ThreadPoolSocketGroup 在 server 的端口上监听收到的消息请求,并交由相应的工作队列处理。每个节点有且仅有一个线程。 BackgroundRoutineService server 的后台线程,每个节点会生成 4 个该线程。server 会在该线程中注册一些函数,这些函数会在BackgroundRoutineService 线程运行过程中每隔一段时间就被调用一次。 LogWriter 将节点运行过程中生成的 log 写入文件的线程。每个节点都有一个该线程。 StdConsole 启动 server 后在命令行窗口接收命令的线程。在 server 启动参数中如果设置 console=true,那么就会启动一个该线程。

优秀的个人博客,低调大师

MongoDB 5.0 正式发布,新增时序集合功能

MongoDB 5.0 正式发布,该版本部分更新内容如下: 时间序列集合 MongoDB 5.0 引入了时间序列集合,它有效地存储了一段时期内的测量序列。与普通集合相比,在时间序列集合中存储时间序列数据可以提高查询效率,并减少数据和索引的磁盘使用。 聚合 新的聚合操作符 MongoDB 5.0 引入了以下聚合操作: $count: $count 在现有管道 $group 阶段和新的 MongoDB 5.0 $setWindowFields 阶段中使用时,提供所有文档的计数。 $dateAdd: 将一个 Date 对象按指定的时间单位递增。 $dateDiff: 返回两个日期之间的差异。 $sampleRate: 添加 $sampleRate 方法,以给定的速度从管道中概率性地选择文档。 $rand: 每次调用 $rand 方法时,都会生成一个0到1之间的随机浮点数。新的 $sampleRate 操作是基于 $rand 的。 窗口操作符 MongoDB 5.0 引入了 $setWindowFields 管道阶段,允许你在一个集合中的指定跨度的文档上执行操作,称为窗口。该操作根据所选择的窗口操作符返回结果。 一般聚合的改进 $expr 操作符:比较运算符使用索引 从 MongoDB 5.0 开始,放在 $expr 操作符中的 $eq、 $lt、 $lte、 $gt和 $gte 操作符可以使用索引来提高性能。 $ifNull 表达式接受多个输入表达式 从 MongoDB 5.0 开始,在返回一个替换表达式之前,你可以为 $ifNull 表达式指定多个输入表达式。 聚合的 let 选项 从 MongoDB 5.0 开始, aggregate 命令和 db.collection.aggregate() 辅助方法有一个 let 选项,用于指定一个可以在聚合管道的其他地方使用的变量列表。这允许你通过将变量与查询文本分开来提高命令的可读性。 $lookup 阶段:简洁的相关子查询 从 MongoDB 5.0 开始,聚合管道 $lookup 阶段支持简明的相关子查询,改善集合之间的连接。 变更事件输出 从 MongoDB 5.0 开始,变更事件包含字段 updateDescription.truncatedArrays 来记录数组的截断。 索引 删除了 geoHaystack 索引和 geoSearch 命令 MongoDB 5.0 删除了废弃的 geoHaystack 索引和 geoSearch 命令。使用带有 $geoNear 的 2d 索引或支持的 geospatial 查询操作符来代替。 将你的 MongoDB 实例升级到 5.0 并将 featureCompatibilityVersion 设置为 5.0 将删除任何先前存在的 geoHaystack 索引。 新的错误信息 db.collection.createIndex() 和 db.collection.createIndexes() 操作在选项指定不正确时有新的错误信息。 中断索引构建 如果副本集中的一个节点在索引构建过程中完全关闭或回滚,索引构建的进度现在会被保存到磁盘上。当服务器重新启动时,索引创建将从保存的位置恢复。 reIndex行为改变 从 MongoDB 5.0 开始,[reIndex](<https://docs.mongodb.com/v5.0/reference/command/reIndex/#mongodb-dbcommand-dbcmd.reIndex>)命令和[db.collection.reIndex()](<https://docs.mongodb.com/v5.0/reference/method/db.collection.reIndex/#mongodb-method-db.collection.reIndex>)shell 方法只能在独立实例上运行。 安全 支持配置 TLS 1.3 密码套件 MongoDB 5.0引入了opensslCipherSuiteConfig参数,以便在使用TLS 1.3加密时,能够配置OpenSSL所支持的密码套件。 完整文档可查看:https://docs.mongodb.com/v5.0/release-notes/5.0/

优秀的个人博客,低调大师

QuestDB —— 高性能时序数据库

QuestDB 是用于时间序列数据的高性能开源 SQL 数据库。它使用面向列的方法、大量并行向量化执行、SIMD 指令以及一系列低延迟技术。整个代码库是从头开始构建的,没有依赖关系,并且 100% 没有垃圾回收。QuestDB 实现了SQL,并使用本地扩展对其进行了时间序列扩展。 它公开了 PostgreSQL 有线协议、高性能 REST API,并支持InfluxDB Line Protocol 的提取。QuestDB 使用具有免维护方案的关系模型。关系和时间序列联接使随时间推移的数据关联变得容易。写入会持久地提交到磁盘,这意味着数据是安全的,但可以立即访问。 性能表现 每个线程每秒的操作数。写入是持久的,并已写入磁盘。在具有 6 个内存通道的 CPU 上,QuestDB 可以每秒扫描 117GB 的数据 使用 96 个可用线程中的 16个 在 c5.metal 实例上的执行时间

优秀的个人博客,低调大师

Pandas时序数据处理入门

图片来源:https://pixabay.com/ 作为一个几乎每天与时间序列数据打交道的人员,我发现pandaPython包在时间序列的操作和分析方面有强大优势。 这篇关于panda时间序列数据处理的基本介绍可以带你入门时间序列分析。本文将主要介绍以下操作: 创建一个日期范围 处理时间戳数据 将字符串数据转换为时间戳 在数据框中索引和切片时间序列数据 重新采样不同时间段的时间序列汇总/汇总统计数据 计算滚动统计数据,如滚动平均值 处理丢失数据 了解unix/epoch时间的基础知识 了解时间序列数据分析的常见陷阱 接下来我们一起步入正题。如果想要处理已有的实际数据,你可能考虑从使用panda read_csv将文件读入数据框开始,然而在这里,我们将直接从处理生成的数据开始。 首先导入我们将会使用到的库,然后用它们创建日期范围 import pandas as pd from datetime import datetime import numpy as np date_rng = pd.date_range(start='1/1/2018', end='1/08/2018', freq='H') 这个日期范围的时间戳为每小时一次。如果我们调用date_rng,我们会看到如下所示: DatetimeIndex(['2018-01-01 00:00:00', '2018-01-01 01:00:00', '2018-01-01 02:00:00', '2018-01-01 03:00:00', '2018-01-01 04:00:00', '2018-01-01 05:00:00', '2018-01-01 06:00:00', '2018-01-01 07:00:00', '2018-01-01 08:00:00', '2018-01-01 09:00:00', ... '2018-01-07 15:00:00', '2018-01-07 16:00:00', '2018-01-07 17:00:00', '2018-01-07 18:00:00', '2018-01-07 19:00:00', '2018-01-07 20:00:00', '2018-01-07 21:00:00', '2018-01-07 22:00:00', '2018-01-07 23:00:00', '2018-01-08 00:00:00'], dtype='datetime64[ns]', length=169, freq='H') 我们可以检查第一个元素的类型: type(date_rng[0]) #returns pandas._libs.tslib.Timestamp 让我们用时间戳数据的创建一个示例数据框,并查看前15个元素: df = pd.DataFrame(date_rng, columns=['date']) df['data'] = np.random.randint(0,100,size=(len(date_rng))) df.head(15) 示例数据框 如果想进行时间序列操作,我们需要一个日期时间索引。这样一来,数据框便可以在时间戳上建立索引。 将数据框索引转换为datetime索引,然后显示第一个元素: df['datetime'] = pd.to_datetime(df['date']) df = df.set_index('datetime') df.drop(['date'], axis=1, inplace=True) df.head() 如果数据中的“时间”戳实际上是字符串类型和数值类型相比较,该怎么办呢?我们可以将date_rng转换为字符串列表,然后将字符串转换为时间戳。 string_date_rng = [str(x) for x in date_rng] string_date_rng #returns ['2018-01-01 00:00:00', '2018-01-01 01:00:00', '2018-01-01 02:00:00', '2018-01-01 03:00:00', '2018-01-01 04:00:00', '2018-01-01 05:00:00', '2018-01-01 06:00:00', '2018-01-01 07:00:00', '2018-01-01 08:00:00', '2018-01-01 09:00:00',... 可以通过推断字符串的格式将其转换为时间戳,然后查看这些值: timestamp_date_rng = pd.to_datetime(string_date_rng, infer_datetime_format=True) timestamp_date_rng #returns DatetimeIndex(['2018-01-01 00:00:00', '2018-01-01 01:00:00', '2018-01-01 02:00:00', '2018-01-01 03:00:00', '2018-01-01 04:00:00', '2018-01-01 05:00:00', '2018-01-01 06:00:00', '2018-01-01 07:00:00', '2018-01-01 08:00:00', '2018-01-01 09:00:00', ... '2018-01-07 15:00:00', '2018-01-07 16:00:00', '2018-01-07 17:00:00', '2018-01-07 18:00:00', '2018-01-07 19:00:00', '2018-01-07 20:00:00', '2018-01-07 21:00:00', '2018-01-07 22:00:00', '2018-01-07 23:00:00', '2018-01-08 00:00:00'], dtype='datetime64[ns]', length=169, freq=None) 但是如果需要转换一个唯一的字符串格式呢? 我们可以创建一个任意的字符串形式的日期列表,并将它们转换为时间戳: string_date_rng_2 = ['June-01-2018', 'June-02-2018', 'June-03-2018'] timestamp_date_rng_2 = [datetime.strptime(x,'%B-%d-%Y') for x in string_date_rng_2] timestamp_date_rng_2 #returns [datetime.datetime(2018, 6, 1, 0, 0), datetime.datetime(2018, 6, 2, 0, 0), datetime.datetime(2018, 6, 3, 0, 0)] 如果把它放到数据框中,将会如何? df2 = pd.DataFrame(timestamp_date_rng_2, columns=['date']) df2 回到最初的数据框架,让我们通过解析时间戳索引来查看数据: 假设只想查看本月2号的数据,可以使用如下索引。 df[df.index.day == 2] 顶部如图所示: 也可以通过数据框索引直接调用想查看的日期: df['2018-01-03'] 如何在特定日期之间选择数据? df['2018-01-04':'2018-01-06'] 我们填充的基本数据框提供了频率以小时计的数据,但同样可以以不同的频率重新采样数据,并指定如何计算新样本频率的汇总统计信息。我们可以取每天频率下数据的最小值、最大值、平均值、总和等,而不是每小时的频率,如下面的例子,计算每天数据的平均值: df.resample('D').mean() 那么诸如滚动平均值或滚动和之类的窗口统计信息呢? 让我们在原来的df中创建一个新列,计算3个窗口周期内的滚动和,然后查看数据框的顶部: df ['rolling_sum'] = df.rolling(3).sum() df.head(10) 可以看到,在这个正确的计算中,只有当存在三个周期可以回顾时,它才开始具有有效值。 这可以有效地帮我们了解到,当处理丢失的数据值时,如何向前或向后“滚动”数据。 这是我们的df,但有一个新的列,采取滚动求和并向后“滚动”数据: df['rolling_sum'] = df.rolling(3).sum() df.head(10) 采用诸如平均时间之类的实际值用于填补丢失的数据,这种方法通常来说是有效的。但一定谨记,如果你正处理一个时间序列的问题,并且希望数据是切合实际的,那么你不应该向后“滚动”数据。因为这样一来,你需要的关于未来的信息就永远不可能在那个时间获取到。你可能更希望频繁地向前“滚动”数据,而不是向后“滚动”。 在处理时间序列数据时,可能会遇到Unix时间中的时间值。Unix时间,也称为Epoch时间,是自协调世界时(UTC) 1970年1月1日星期四00:00:00以后经过的秒数。使用Unix时间有助于消除时间戳的歧义,这样我们就不会被时区、夏令时等混淆。 下面是一个时间t在Epoch时间的例子,它将Unix/Epoch时间转换为UTC中的常规时间戳: epoch_t = 1529272655 real_t = pd.to_datetime(epoch_t, unit='s') real_t #returns Timestamp('2018-06-17 21:57:35') 如果我想把UTC中的时间转换为自己的时区,可以简单地做以下操作: real_t.tz_localize('UTC').tz_convert('US/Pacific') #returns Timestamp('2018-06-17 14:57:35-0700', tz='US/Pacific') 掌握了这些基础知识后,就可以开始处理时间序列数据了。 以下是一些处理时间序列数据时要记住的技巧和常见的陷阱: 检查数据中可能由区域特定时间变化(如夏令时)引起的差异。 精心跟踪时区- 让他人通过代码了解你的数据所在的时区,并考虑转换为UTC或标准化值以保持数据标准化。 丢失的数据可能经常发生 - 请确保记录清洁规则并考虑不回填在采样时无法获得的信息。 请记住,当重新采样数据或填写缺失值时,将丢失有关原始数据集的一定数量的信息。建议跟踪所有数据转换并跟踪数据问题根源。 重新采样数据时,最佳方法(平均值,最小值,最大值,总和等)取决于拥有的数据类型以及采样方式。请仔细考虑如何重新采样数据以进行分析。 Programming Python Pandas Timeseries Data Science 以上为译文 本文由阿里云云栖社区组织翻译。 文章原标题《Basic Time Series Manipulation with Pandas》,译者:狮子家的袋鼠,审校:么凹。 原文链接​ 本文为云栖社区原创内容,未经允许不得转载。

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长,特征是大鼻子、头戴帽子、身穿背带裤,还留着胡子。与他的双胞胎兄弟路易基一起,长年担任任天堂的招牌角色。

腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题,腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构,目前腾讯云软件源站支持公网访问和内网访问。

Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能,例如代码缩略图,Python的插件,代码段等。还可自定义键绑定,菜单和工具栏。Sublime Text 的主要功能包括:拼写检查,书签,完整的 Python API , Goto 功能,即时项目切换,多选择,多窗口等等。Sublime Text 是一个跨平台的编辑器,同时支持Windows、Linux、Mac OS X等操作系统。

WebStorm

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源,继承了IntelliJ IDEA强大的JS部分的功能。

用户登录
用户注册