首页 文章 精选 留言 我的

精选列表

搜索[笔记],共6930篇文章
优秀的个人博客,低调大师

《老男孩Linux运维笔记笔记

centos7为基准 1、更改YUM源: /bin/mv CentOS-Base.repo CentOS-Base.repo.ori //备份 yum 源 7 wget http://mirrors.sohu.com/help/CentOS-Base-sohu.repo //下载soho 源 /bin/mv CentOS-Base-sohu.repo CentOS-Base.repo 打补丁: yum upgrade 安装必要的软件包: yum -y install lrzsz 2、关闭selinux: vi /etc/sysconfig/selinux disabled 重启 3、设定运行级别: systemctl get-default systemctl set-default shutdown.target(0) emergency.target rescure.target multi-user.target(3) graphical.target(5) 4、关闭不必要的服务 systemctl list-units --type=service 开机启动:systemctl enable crond network sshd syslog 5、授权服务:visudo root ALL=(ALL) ALL %wheel ALL=(ALL) ALL %wheel ALL=(ALL) NOPASSWD: ALL 6、ssh优化:vi /etc/ssh/sshd_config 52113#→ssh 连接默认的端口,谁都知道,必须要改。 PermitRootLogin no#→root 用户黑客都知道的,禁止它远程登陆。 PermitEmptyPasswords no #→禁止空密码登陆 UseDNSno#→不使用DNS GSSAPIAuthentication no 重启sshd 服务#/etc/init.d/sshd restart 7、修改默认字符集: locale -a vi /etc/locale.conf source /etc/locale.conf 8、同步时间服务: echo '*/5 * * * * /usr/sbin/ntpdate time.nist.gov >/dev/null 2>&1' >>/var/spool/cron/root 9、修改文件描述符: ulimit -n echo '* - nofile 65535 ' >>/etc/security/limits.conf 10、清理clientmqueue垃圾文件防止inode被占满 find /var/spool/clientmqueue/ -type -f |xargs rm -f 11、内核优化:vi /etc/sysctl.conf net.ipv4.tcp_fin_timeout = 2 net.ipv4.tcp_tw_reuse = 1 net.ipv4.tcp_tw_recycle = 1 net.ipv4.tcp_syncookies = 1 net.ipv4.tcp_keepalive_time = 600 net.ipv4.ip_local_port_range = 4000 65000 net.ipv4.tcp_max_syn_backlog = 16384 net.ipv4.tcp_max_tw_buckets = 36000 net.ipv4.route.gc_timeout = 100 net.ipv4.tcp_syn_retries = 1 net.ipv4.tcp_synack_retries = 1 net.core.somaxconn = 16384 12 net.core.netdev_max_backlog = 16384 net.ipv4.tcp_max_orphans = 16384 net.ipv4.ip_conntrack_max = 25000000 net.ipv4.netfilter.ip_conntrack_max=25000000 net.ipv4.netfilter.ip_conntrack_tcp_timeout_established=180 net.ipv4.netfilter.ip_conntrack_tcp_timeout_time_wait=120 net.ipv4.netfilter.ip_conntrack_tcp_timeout_close_wait=60 net.ipv4.netfilter.ip_conntrack_tcp_timeout_fin_wait=120 12、grep高亮显示: vi /etc/profile alias grep='grep --color=auto' source /etc/profile 13、锁定关键文件 chattr +i /etc/passwd chattr +i /etc/inittab chattr +i /etc/group chattr +i /etc/shadow chattr +i /etc/gshadow 14、更改登录后的信息:/etc/motd 15、内核参数: 16、NFS:yum install -y nfs-utils rpcbind vi /etc/exports xx ip(rw) exportfs -r mount -t nfs xx:/xx /yy mount -t nfs -o nosuid,noexec,nodev,rw,bg,soft,rsize=32768,wsize=32768 192.168.1.4:/home/test /usr/local/live/ 17、apache安装调优参数: ./configure \ --prefix=/application/apache2.2.27 \#安装的目录 --enable-deflate \ #压缩文件文本一般 html/js/css 等内容的站点,使用此参数功能会大大提高传输速度,提示访问者访问体验,在生产环境中,这是apache 调优的重要选项之一 --enable-expires \ #激活允许通过配置文件控制http 的“expires”和“cache-control” 头内容,即对网站图片,js,css 等内容,提供在客户端浏览器缓存的设置,这是apache 调优的重 要选项之一 --enable-headers \ #提供允许http 请求头的控制 --enable-moudles=most \ #激活多数模块 --enable-so \ #激活apache 服务的DSO 支持,即在以后可以以DSO 的方式编译 安装共享模块,这个模块本身不能以DSO 方式编译 --with-mpm=worker \ #选择apache mpm 的模式为worker 模式,因为worker 模式原 理是更多的使用线程来处理请求,所以可以处理更多的并发请求,而系统资源的开销小于基于进程 的MPM prefork,如果不指定此参数,默认得到模式是prefork 进程模式。这是apache 调优的一 个重要选项之一。 --enable-rewrite #提供基于URL 规则的重写功能,根据已知URL 地址,转换其他想 要的url 地址,如伪静态功能就是这个模块实现的,这是apache 在生产环节中必用的一个重要功能 检查语法:apachectl -t 启用: apachectl start 平滑重启:apachectl graceful 18、apache配置文件:Httpd.conf ServerRoot "/usr/local/xxx" #安装目录 Listen 80 <ifModule !mpm_netware_moudle> User daemon #执行者 Group daemon ServerAdmin xx@qq.com DocumentRoot "/xx" #网站根目录 #设置根目录访问权限 <Directory /> Options FollowSymLinks AllowOverride None Order allow,deny Allow from all </Direcotry> <ifModule dir_module> DirectoryIndex index.html </ifModule> #AllowOverride:允许存在.htaccess 文件中的指令类型None:当AllowOverride 设置None 时,不搜索该目录下的.htaccessAll:在.htaccess 文件中使用所有指令 #Allow:允许访问的主机列表(可用域名或子网,例如:Allow from 192.168.0.0/16)。 #DirectoryIndex index.html index.htm index.php #主页文件的设置(本例将主页文件设置为: index.html,index.htm 和index.php) #Options特性:ExecCGI 在该目录下执行CGI脚本FollowSymLinks:在该目录下允许文件系统使用符号连接Indexes: 当用户访问该目录时,如果用户找不到DirectoryIndex 指定的主页文件(例如index.html),则返回该目录下的文件列表给用户。SymLinksIfOwnerMatch: 当使用符号连接时,只有当符号连接的文件拥有者与实际文件的拥有者相同时才可以访问。 #设置其他目录权限 <Direcotry "/usr/local"> Options Indexes FollowSymLinks AllowOverride None Order allow,deny Allow from all 19、apache的PHP.INI配置 /application/php/lib/php.ini [PHP] engine = On ——→ 是否启用PHP 解析引擎 short_open_tag = Off ——→ 是否使用简介标志 asp_tags = Off ——→ 不允许asp 类标志 precision = 14 ——→ 浮点型数据显示的有效期 y2k_compliance = On output_buffering = 4096 ——→ 输出缓冲区大小(字节)。建议值为4096~8192。 zlib.output_compression = Off ——→ 是否开启zlib 输出压缩 implicit_flush = Off ——→ 是否要求PHP 输出层在每个输出块之后自动刷新数据 这等效于在每个print()、echo()、HTML 块之后自动调用flush()函数。打开这个选项对程序执行 的性能有严重的影响,通常只推荐在调试时使用。在CLI SAPI 的执行模式下,该指令默认为On 。 serialize_precision = 17 safe_mode = Off ——→ 安全模式 safe_mode_gid = Off safe_mode_exec_dir = ——→ 安全模式下的可执行文件存放目录 safe_mode_allowed_env_vars = PHP_ ####在安全模式下,用户仅可以更改的环境变量的前缀列表(逗号分隔)。允许用户设置某些环境变 量,可能会导致潜在的安全漏洞。注意: 如果这一参数值为空,PHP 将允许用户更改任意环境变量。 safe_mode_protected_env_vars = LD_LIBRARY_PATH ####在安全模式下,用户不能更改的环境变量列表(逗号分隔)。这些变量即使在safe_mode_allowed_env_vars 指令设置为允许的情况下也会得到保护。 disable_functions = ——→ 该指令接受一个用逗号分隔的函数名列表,以禁用特定的函数。 disable_classes = ——→ 该指令接受一个用逗号分隔的类名列表,以禁用特定的类 expose_php = On ——→ 在网页头部显示php 信息 max_execution_time = 30 ——→ 每个脚本最大执行秒数 max_input_time = 60 ——→ 每个脚本用来分析请求数据的最大限制时间 memory_limit = 128M ——→ 每个脚本执行的内存限制 72 display_errors = Off ——→ #显示失误(该关闭,换成日志显示) display_startup_errors = Off ——→ #显示启动失误 log_errors = On ——→ 生成错误错误日志显示 log_errors_max_len = 1024 ——→ 设定error_log 最大长度 ignore_repeated_errors = Off ——→ 打开后,不记录重复的信息 ignore_repeated_source = Off ——→ 打开后当记录重复的信息时忽略来源 report_memleaks = On ——→ 报告内存泄露,仅在debug 编译模式下有效 html_errors = Off ——→ 是否开启静态网页错误提示 register_globals = Off ——→ ##是否打开register 全局变量 register_long_arrays = Off ####是否注册老形式的输入数组, HTTP_GET_VARS 和相关数组;如果你不使用他们,建议为了提 高性能关闭他们. register_argc_argv = Off ####此指令让PHP 确认是否申明argv&argc 变量(这些变量会包含GET 信息). ;如果你不使用这 些变量,为了提升性能应该关闭此选项. auto_globals_jit = On ####当打开此项, SERVER 和 ENV 变量将在第一次被使用时而不是脚本一开始时创建(运行时);如 73 果这些变量在脚本中没有被使用过, 打开此项会增加一点性能.;为了使此指令有效,PHP 指令 register_globals, register_long_arrays,;以及register_argc_argv 必须被关闭. post_max_size = 8M ——→ #PHP 可以接受的最大的POST 数据大小 magic_quotes_sybase = Off ##使用Sybase 风格的magic quotes (使用"来引导'替代\'). auto_prepend_file = ——→ #在任何PHP 文档之前或之后自动增加文件 auto_append_file = ####两个有趣的变量是auto_prepend_file 以及auto_append_file。这些变量指定PHP 自动添加 到任何PHP 文档文件头或文件尾的其他文件。这对于为PHP 产生的页面添加页眉或页脚非常有用, 可以节省为每个PHP 文档添加代码的时间。但需要注意这里的指定文件将会添加到所有的PHP 文 档中,所以这些变量必须适合单应用程序(single-application)的服务器。所包含的文件要么是 PHP 脚本,要么是普通的HTML 文档。嵌入式PHP 代码必须用标准标记括起来。 default_mimetype = "text/html" ——→ #PHP 内建默认为text/html doc_root = ——→ #PHP 的"根目录"。仅在非空时有效。 file_uploads = On 是否开启上传功能 upload_max_filesize = 2M #最大可上传文件,2M max_file_uploads = 20 最大同时可以上传20 个文件 allow_url_fopen = On #是否允许打开远程文件 allow_url_include = Off #是否允许include/require 远程文件 default_socket_timeout = 60 默认的socket 超时时间 pdo_mysql.cache_size = 2000 ——→ Ped_mysql 的缓存大小 pdo_mysql.default_socket= ——→ 默认的socket 时间 [Phar] [Syslog] define_syslog_variables = Off ——→ 是否定义各种的系统日志变量 [mail function] ——→ 邮件功能 SMTP = localhost ——→ 本地作为邮件服务器 smtp_port = 25 邮件端口号默认是25 mail.add_x_header = On ——→ 是否开启最大的header [ODBC] odbc.allow_persistent = On ——→ 允许或阻止持久连接. odbc.check_persistent = On ——→ 在重用前检查连接是否可用 odbc.max_persistent = -1 ——→ 持久连接的最大数目,-1 意味着没有限制. odbc.max_links = -1 ——→ 最大连接数(持久+ 非持久).-1 意味着没有限制. odbc.defaultlrl = 4096 ——→ 长字段处理.返回变量的字节数.0 意味着略过. odbc.defaultbinmode = 1 ####二进制数据处理.0 意味着略过,1 按照实际返回,2 转换到字符.;查看odbc_binmode 和 odbc_longreadlen 的文档来获取针对uodbc.defaultlrl 和uodbc.defaultbinmode 的解释 [Interbase] ——→ Interbase 数据库 75 ibase.allow_persistent = 1 ——→ 允许或组织持久连接。 ibase.max_persistent = -1 ——→ 持久连接的最大数目,-1 意味着没有限制. ibase.max_links = -1 ——→ 最大连接数(持久+ 非持久).-1 意味着没有限制. ibase.timestampformat = "%Y-%m-%d %H:%M:%S" ——→ 数据库时间记录模式 ibase.dateformat = "%Y-%m-%d" ibase.timeformat = "%H:%M:%S" [MySQL] mysql.allow_local_infile = On ——→ 是否允许本地文件连接数据库 mysql.allow_persistent = On ——→ 允许或禁止持久连接 mysql.cache_size = 2000 ——→ mysql 缓存大小 mysql.max_persistent = -1 ——→ 持久连接的最大数目. -1 意味着没有限制. mysql.max_links = -1 ——→ 连接的最大数目(持久和非持久)。-1 代表无限制 mysql.default_port = ####mysql_connect() 使用的默认端口,如不设置,mysql_connect() ;将使用变量$MYSQL_TCP_PORT,或在/etc/services 下的mysql-tcp 条目(unix), ;或在编译是定义的MYSQL_PORT(按这样的顺序) mysql.default_socket = ####用于本地MySql 连接的默认的套接字名。为空,使用MYSQL 内建值 mysql.default_host = ——→ mysql_connect() 默认使用的主机(安全模式下无效) mysql.default_user = ——→ mysql_connect() 默认使用的用户名(安全模式下无效) mysql.default_password = ——→ mysql_connect() 默认使用的密码(安全模式下无效 mysql.connect_timeout = 60 ——→ 连接超时时间,默认是60s mysql.trace_mode = Off [MySQLi] mysqli.max_persistent = -1 ——→ 持久连接的最大数目. -1 意味着没有限制. mysqli.allow_persistent = On ——→ 允许或拒绝之久连接 76 mysqli.max_links = -1 ——→ 最大连接数. -1 意味着没有限制. mysqli.cache_size = 2000 ——→ 连接缓存大小 mysqli.default_port = 3306 ——→ 连接端口号 ####mysqli_connect()默认的端口号.如果没有设置, mysql_connect() 会使用 $MYSQL_TCP_PORT;或者位于/etc/services 的mysql-tcp 入口或者编译时定义的 MYSQL_PORT 值(按照此顺序查找).;Win32 只会查找MYSQL_PORT 值. mysqli.default_socket = ####对于本地MySQL 连接的默认socket 名称. 如果为空, 则使用MySQL 内建默认值. mysqli.default_host = ####mysqli_connect()的默认host 值(在安全模式中不会生效) mysqli.default_user = ####mysqli_connect()的默认user 值(在安全模式中不会生效). mysqli.default_pw = ####mysqli_connect() 的默认password 值(在安全模式中不会生效). ; 注意在此文件中保存密码一般来说是*糟糕* 的主义. ; *任何* 使用PHP 的用户可以执行'echo get_cfg_var("mysqli.default_password") ; 并且获取到此密码! 而且理所当然, 任何有对此文件读权限的用户都可以获取到此密码. mysqli.reconnect = Off ——→ 允许或阻止持久连接 [mysqlnd] mysqlnd.collect_statistics = On mysqlnd.collect_memory_statistics = Off [OCI8] [PostgreSQL] pgsql.allow_persistent = On ——→ 允许或阻止持久连接. pgsql.auto_reset_persistent = Off ####总是在pg_pconnect() 时检测断开的持久连接.;自动重置特性会引起一点开销. pgsql.max_persistent = -1 ——→ 持久连接的最大数目. -1 意味着没有限制. 77 pgsql.max_links = -1 ——→ 最大连接数(持久+ 非持久). -1 意味着没有限制 pgsql.ignore_notice = 0 ——→ 是否忽略PostgreSQL 后端通告消息.;通告消息记录会需要 一点开销. pgsql.log_notice = 0 ####是否记录PostgreSQL 后端通告消息.;除非pgsql.ignore_notice=0, 否则模块无法记录通 告消息。 [Sybase-CT] sybct.allow_persistent = On ——→ 允许或阻止持久连接. sybct.max_persistent = -1 ——→ 持久连接的最大数目. -1 意味着没有限制. sybct.max_links = -1 ——→ 最大连接数(持久+ 非持久). -1 意味着没有限制. sybct.min_server_severity = 10 ——→ 显示出的错误最小严重程度. sybct.min_client_severity = 10 ——→ 显示出的消息最小严重程度 [bcmath] bcmath.scale = 0 ——→ #用于所有bcmath 函数的10 十进制数数字的个数 [browscap] [Session] session.save_handler = files ——→ 用于保存/取回数据的控制方式 session.use_cookies = 1 ——→ 是否使用cookies session.use_only_cookies = 1 ####这个选项允许管理员去保护那些在URL 中传送session id 的用户免于被攻击;默认是0. session.name = PHPSESSID ——→ session 的名字(同时作为cookie 的名称 session.auto_start = 0 ——→ 在请求开始时初始化session session.cookie_lifetime = 0 ——→ cookie 的存活秒数,如果为0,则是直到浏览器重新启动 session.cookie_path = / ——→ cookie 的有效路径 session.cookie_domain = ——→ cookie 的有效域名 session.cookie_httponly = ####是否将httpOnly 标志增加到cookie 上,增加后则 cookie 无法被浏览器的脚本语言(例如 JavaScript)存取. 78 session.serialize_handler = php 用于序列化数据的处理器. php 是标准的PHP 序列化器. session.gc_probability = 1 ####; 定义'垃圾回收'进程在每次session 初始化时开始的比例. ; 比例由gc_probability/gc_divisor 来得出, ; 例如. 1/100 意味着在每次请求时有1%的机会启动'垃圾回收'进程. session.gc_divisor = 1000 session.gc_maxlifetime = 1440 ####在这里数字所指的秒数后,保存的数据将被视为'碎片(garbage)'并由gc 进程清理掉。 session.bug_compat_42 = Off ####PHP 4.2 和更早版本有一个未公开的特性/bug , 此特性允许你在全局初始化一个session 变量,即便register_globals 已经被关闭.;如果此特性被使用,PHP 4.3 和更早版本会警告你.;你可以 关闭此特性并且隔离此警告. 这时候,如果打开bug_compat_42,那此警告只是被显示出来. session.bug_compat_warn = Off session.referer_check = ####检查HTTP Referer 来防止带有id 的外部URL.;HTTP_REFERER 必须包含从session 来的这 个字段才会被认为是合法的. session.entropy_length = 0 ——→ 从此文件读取多少字节 session.cache_limiter = nocache ####设置为{nocache,private,public,}来决定HTTP 缓冲的类型;留空则防止发送anti-caching 头. session.cache_expire = 180 ——→ 文档在n 分钟之后过期. session.use_trans_sid = 0 ####trans sid 支持默认关闭. ;使用trans sid 可能让你的用户承担安全风险.;使用此项必须小心.; - 用户也许通过email/irc/其他 途径发送包含有效的session ID 的URL 给其他人.; - 包含有效session ID 的URL 可能被存放在容 易被公共存取的电脑上.; - 用户可能通过在浏览器历史记录或者收藏夹里面的包含相同的session ID 的URL 来访问你的站点. 20、httpd-mpm.conf文件详解 #prefork 多路处理模块 <IfModule mpm_prefork_module> StartServers 5 #设置服务器启动时建立的子进程数量,一般不调 MinSpareServers 5 #设置空闲子进程的最小数量,不要调太大 MaxSpareServers 10 #设置空闲子进程的最大数量 MaxClients 150 #用于服务器客户端最大请求数量 MaxRequestsPerChild 0 #每个子进程在生存期内允许服务器的最大请求数,建议10000-30000 #worker 多路处理模块 <IfModule mpm_worker_module> StartServers 2 #设置服务器启动时建立的子进程数量,一般不调 MaxClients 150 #用于服务器客户端最大请求数量 MinSpareThreads 25 #设置空闲子进程的最小数量,不要调太大 MaxSpareThreads 75 #设置空闲子进程的最大数量 ThreadsPerChild 25 #每个子进程建立的线程数 MaxRequestsPerChild 0 #设置每个子进程在其生存期内允许伺服的最大请求数量 21、http-default.conf详解 Timeout 300 #设置服务器在断定请求失败前等待的秒数。默认值300 KeepAlive Off #设置是否启用HTTP 持久链接,On 代表打开,Off 代表关闭。 如果用于同一页面包含大量静态文件的应用,设置为On,以提高性能; 如果用于主要为动态页面的应用,设置为Off,以节约内存资源; 如果服务器前跑有squid 或者其它七层设备,设置为On MaxKeepAliveRequests 100 #限制当启用KeepAlive 时,每个连接允许的请求数量。 如果将此值设为"0",将不限制请求的数目。 笔者建议将此值设为100-500 之间的一个值, 以确保最优的服务器性能 KeepAliveTimeout 5 #设置持久链接中服务器在两次请求之间等待的秒数。对于高负荷服务器来说, KeepAliveTimeout 值较大会导致一些性能方面的问题: 超时值越大,与空闲客户端保持连接的进程就越多 UseCanonicalName Off #配置服务器如何确定它自己的域名, 可选值为On | Off | DNS。DNS 用于为大量基于IP 的虚拟主机支持那些古董级的不提供"Host:"头的浏览器使 用。 笔者建议设置为Off AccessFileName .htaccess #设置分布式配置文件的名字,默认为.htaccess。 如果为某个目录启用了分布式配置文件功能,那么在向客户端返回其中的文档时, 服务器将在这个文档所在的各级目录中查找此配置文件,因此会带来性能问题, 笔者建议关闭分布式配置文件功能。 ServerTokens Prod #控制服务器回应给客户端的"Server:"应答头是否包含关于服务器操作系统类型和编译 进的模块描述信息, 同时还控制着ServerSignature 指令的显示内容。可选值为Full | OS | Minor | Minimal | Major | Prod。 笔者建议设置为显示最少信息的Prod。 ServerSignature Off #配置服务器生成页面的页脚,可选值为On | Off | EMail。 采用On 会简单的增加一行关于服务器版本和正在伺服的虚拟主机的ServerName, 而EMail 设置会额外创建一个指向ServerAdmin 的"mailto:"部分。建议使用默认值Off。 HostnameLookups Off #设置是否启用对客户端IP 的DNS 查找,可选值为On | Off | Double。 DNS 查询会造成明显的时间消耗,建议设置为Off。 22、bin目录 23、虚拟主机配置: vim httpd-vhosts.conf NameVirtualHost *:80 <VirtualHost *:80> ServerAdmin 291406980@qq.com DocumentRoot "/var/www/html/www" ServerName www.etiantian.org ServerAlias etiantian.org ErrorLog "logs/www-error_log" CustomLog "logs/www-access_log" common vim httpd.conf Include conf/extra/httpd-vhosts.conf #基于端口 NameVirtualHost *:80 NameVirtualHost *:8000 NameVirtualHost *:9000 #基于IP <VirtualHost 192.168.1.1:90> xxxxxxxxx 24、apache日志: CustomLog "logs/blog-access_log" combined #日志轮询 #统计IP awk '{print $1}' access_bbs_xx.log|sort|uniq -c|sort -rn 25、隐藏版本信息: 方法一: ※首先修改源文件,再进行make && make install 编译安装 编辑源文件/usr/local/apache2/include/ap_release.h 文件 [root@Nagios-Server include]# vimap_release.h #define AP_SERVER_BASEPRODUCT "IIS" #define AP_SERVER_MAJORVERSION_NUMBER 7 #define AP_SERVER_MINORVERSION_NUMBER 0 #define AP_SERVER_PATCHLEVEL_NUMBER 0 #define AP_SERVER_DEVBUILD_BOOLEAN 0 编辑源文件/usr/local/apache2/include/os.h 文件 [root@Nagios-Server include]# vimos.h #define PLATFORM "Win32" 方法二: [root@Nagios-Server include]# vim /usr/local/apache2/conf/httpd.conf # Various default settings Include conf/extra/httpd-default.conf [root@Nagios-Server include]# vim /usr/local/apache2/conf/extra/httpd-default.conf #ServerTokens Prod #ServerSignature off [root@Nagios-Server include]# curl -I 192.168.1.125 89 HTTP/1.1 200 OK Date: Sun, 07 Dec 2014 11:55:51 GMT Server: Apache/2.2.23 (Unix) PHP/5.4.1 #此处无法去掉,若要隐藏,只有用方法一 Last-Modified: Sat, 06 Dec 2014 07:22:37 GMT ETag: "42760-19-509870ed29d1c" Accept-Ranges: bytes Content-Length: 25 Content-Type: text/html 26、apache的rewrite:vi /etc/httpd.conf LoadModule rewrite_module modules/mod_rewrite.so RewriteEngine on Include conf.d/xx.conf RewriteCond %{HTTP_HOST} hunk.test.com [NC] RewriteRule ^(.*)/index.html$ http://hunk.test.com/test.html [L,R=302] 27、apache安全 Allow from [All /全域名/部分域名/IP 地址/网络地址/CIDR 地址] All:表示全部客户端 全域名:表示域名对应的客户端,如www.domain.com 部分域名:表示域名内所有客户端,如domain.com IP 地址:如172.20.17.1 网络地址:如172.20.17.0/255.255.255.0 CIDR 地址:如172.20.17.0/24 htpasswd -c /usr/local/apache2/conf/users sam 28、图片防盗链 28、忽略某些访问日志 29、APACHE日志分析 1.获得访问前10 位的ip 地址 [root@apache ~]# cat access_log |awk '{print $1}'|sort|uniq -c|sort -nr|head -10 2.访问次数最多的文件或页面,取前20 cat access.log|awk ‘{print $11}’|sort|uniq -c|sort -nr|head -20 3.列出传输最大的几个exe 文件 cat access.log |awk ‘($7~/\.exe/){print $10 ” ” $1 ” ” $4 ” ” $7}’|sort -nr|head -20 4. 列出输出大于200000byte(约200kb)的exe 文件以及对应文件发生次 数 cat access.log |awk ‘($10 > 200000 && $7~/\.exe/){print $7}’|sort -n|uniq -c|sort -nr|head -100 5. 如果日志最后一列记录的是页面文件传输时间,则有列出到客户端 最耗时的页面 cat access.log |awk ‘($7~/\.php/){print $NF ” ” $1 ” ” $4 ” ” $7}’|sort -nr|head -100 6. 列出最最耗时的页面(超过60 秒的)的以及对应页面发生次数 cat access.log |awk ‘($NF > 60 && $7~/\.php/){print $7}’|sort -n|uniq -c|sort -nr|head -100 7. 列出传输时间超过30 秒的文件 cat access.log |awk ‘($NF > 30){print $7}’|sort -n|uniq -c|sort -nr|head -20 8. 统计网站流量(G) 94 cat access.log |awk ‘{sum+=$10} END {print sum/1024/1024/1024}’ 9. 统计404 的连接 awk ‘($9 ~/404/)’ access.log | awk ‘{print $9,$7}’ | sort 10. 统计http status. cat access.log |awk ‘{counts[$(9)]+=1}; END {for(code in counts) print code, counts[code]}' cat access.log |awk '{print $9}'|sort|uniq -c|sort -rn 11. 蜘蛛分析 查看是哪些蜘蛛在抓取内容。 /usr/sbin/tcpdump -i eth0 -l -s 0 -w - dst port 80 | strings | grep -i user-agent | grep -i -E 'bot|crawler|slurp|spider' Webalizer 日志分析程序 [root@Apache-Server tools]# yum install -y webalizer [root@Apache-Server tools]# less /etc/webalizer.conf 30、apache tomcat整合 http://blog.51cto.com/maofan/1560639

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

Hive笔记

一、概述 1.1 简介 (1)Hive提供了一个被称为Hive查询语言(简称HiveQL或HQL)的SQL语言,来查询存储在HDFS中的结构化数据文件,它把HQL语句的查询转换为MapReduce任务。 (2)Hive应用场景: a、数据仓库:数据抽取、数据加载、数据转换 b、数据汇总:每天/每周用户点击数、流量统计 c、非实时分析:日志分析、文本分析 d、数据挖掘:用户行为分析、兴趣分区、区域展示 1.2 架构 hive是典型C/S模式,Client端有JDBC/ODBC Client和Thrift Client两类。Server端则分为如下几个部分: CLI: CLI是和Hive交互的最简单/最常用方式,你只需要在一个具备完整Hive环境下的Shell终端中键入hive即可启动服务。 Thrift Server: Hive Thrift Server是基于Thrift 软件框架开发的,它提供Hive的RPC通信接口。目前的HiveServer2(HS2)较之前一版HiveServer,增加了多客户端并发支持和认证功能,极大地提升了Hive的工作效率和安全系数。 Metastore: Metastore是Hive元数据的存储地。在功能上Metastore分为两个部分:服务和存储,也就是架构图中提到的Metastore及其Database。通常使用MySQL来存储元数据。 WUI: WUI并不属于Apache Hive,它是Hive生态圈的一项服务,目前熟知的有Karmasphere、Hue、Qubole等项目。WUI是B/S模式的服务进程,Server一端与Hive Thrfit Server交互,Brower一端供用户进行Web访问。 Driver: 每一个Hive服务都需要调用Driver来完成HQL语句的翻译和执行。通俗地说,Driver就是HQL编译器,它解析和优化HQL语句,将其转换成一个Hive Job(可以是MapReduce,也可以是Spark等其他任务)并提交给Hadoop集群。 1.3 特性 (1)优点: a、高可靠、高容错:HiveServer采用主备模式、双MetaStore、超时重试 b、类SQL:类似sql语句、内置大量函数 c、可扩展:自定义存储格式、自定义函数 (2)缺点: a、延迟较高:默认M/R为执行引擎、M/R启动有延迟 b、不支持物化视图:虽然提供了视图的概念,但还不支持物化视图;不能再视图上更新、插入、删除数据 c、不适用OLTP:暂不支持行级别的数据添加、更新、删除操作 d、暂无支持存储过程 二、数据类型 Hive表字段所支持的数据类型分为两类:基本数据类型和集合数据类型。 2.1 基本数据类型 tinyint、smallint、int、bigint、boolean、float、double、string、timestamp、binary 2.2 集合数据类型 struct(结构体或者对象)、map(键-值对)、array(数组) 举个例子,创建一张员工表。 CREATE TABLE employees ( name STRING, salary FLOAT, subordinates ARRAY<STRING>, deductions MAP<STRING, FLOAT>, address STRUCT<street:STRING, city:STRING, state:STRING, zip:INT>); name表示员工姓名,salary是薪水,subordinates是下属员工的姓名集合数组,deductions是一个由键-值对构成的map,记录了每一次的扣除额,address是家庭住址,使用struct数据类型存储,其中的每个域都被作了命名,并且具有一个特定的类型。 三、读时模式 VS 写时模式 传统型数据库采用写时模式,即在数据写入数据库时对模式进行检查。Hive不会在数据加载时进行验证,而是在查询时进行,也就是读时模式。如果模式和文件内容并不匹配,hive会极力读取这些数据。比如每行记录中的字段个数少于对应的模式中定义的字段个数的话,那么用户将会看到查询结果中多出来的字段值为null。 四、HiveQL:数据定义 4.1 新增数据库 create database [数据库名]; 如果没有显示指定数据库,那么将会使用默认的数据库default; 数据库所在的目录位于属性 hive.metastore.warehouse.dir 所指定的顶层目录之后,假设使用的是默认的配置,也就是/user/hive/warehouse,那么当我们创建数据库financials时,Hive将会对应地创建一个目录/user/hive/warehouse/financials.db,financials.db是一个文件夹,在financials数据库下建的表对应的文件都放这个目录下面。 4.2 建表 4.2.1 管理表(内部表) 默认创建的表即为管理表(内部表),区别于外部表,管理表(内部表)的数据文件是加载到内部,由hive来管理。 create table 表名(列名 列类型,列名 列类型......) location '路径'; location指定表对应的目录,默认为对应数据库目录下面,如在financials数据库下面新建的records表对应的目录为/user/hive/warehouse/financials.db/records,后面该表对应的数据文件会存储到该目录下。 4.2.2 外部表 create external table if not exists [表名] (列名 列类型,列名 列类型.....) row format delimited fields terminated by ',' location '路径'; 关键字external 来表名创建的是外部表,location声明该表对应的数据文件的位置,该位置处于外部系统,hive不会把该数据文件加载到系统内。在删除表时,hive不会同时此数据文件。 外部表适用于hive和其他的工具共用一份数据,可以进行一些查询的操作,但hive并不拥有此数据的所有权。 4.2.3 分区表 创建一张表partition_log,以dt和country这两个字段进行分区。 create table partition_log(ts bigint, line string) partitioned by (dt string, country string); 从本地文件给表partition_log加载数据,在加载时要为加载的数据指定分区,这里指定的值dt为2001-01-01,2001-01-01为GB。 load data local inpath '/root/hive/partitions/file1' into table partition_log partition (dt='2001-01-01', country='GB'); hive此时会在partition_log表对应的路径下面创建 /dt=2001-01-01/country=GB/ 这两级目录。 分区表本质上就是将数据进行归类,同一类的数据放到一个文件夹下,提供查询的效率。 4.2.4 外部分区表 要使用外部分区表,首先要创建一张外部表。 create external table external_table (列名 列类型,列名 列类型....) partitioned by (year int, month int ,day int); 用alter table为表增加一个分区,并为该分区指定一个外部文件的路径。 alter table external_table add partition(year=2012,month=1,day=2) location 'hdfs://master_server/data/log_messages/2012/01/02'; 五、HiveQL:数据操作 5.1 向管理表中装载数据 load data local inpath '${env.HOME}/california-employees' overwrite into table employees partition (country='US', state='CA'); inpath指定数据文件的路径,此路径可以是一个目录,hive会把该目录下所有的文件都加载进来。partition指定分区,如果不是分区可以省略。overwrite关键字表示覆盖,hive会把原先表目录下的数据文件删除再加载新的数据文件。 5.2 通过查询语句向表中插入数据 insert overwrite table employees partition (country='US',state='OR') select * from staged_employees se where se.cntry='US' and se.st='OR'; 如果向多个分区插入数据,可以采用如下方式: from staged_employees se insert overwrite table employees partition(country='US' , state='OR') select * where se.cnty='US' and se.st='OR' insert overwrite table employees partition(country='US',state='CA') select * where se.cnty='US' and se.st='CA'; 动态分区插入 基于查询参数推断出需要创建的分区名称。 insert overwrite table employees partition (country,state) select ... ,se.cnty,se.st from staged_employees se; Hive根据select语句中最后2列来确定分区字段country和state的值。 也可以混合使用动态和静态分区,下面例子指定了country字段的值为静态的US,而分区字段state是动态值: insert overwrite table employees partition (country='US',state) select ...,se.cnty,se.st from staged_employees se where se.cnty='US'; 静态分区键必须出现在动态分区键之前。 5.3 单个查询语句中创建表并加载数据 create table ca_employees as select name,salary,address from employees where se.state='CA'; 六、HiveQL:查询 6.1 select ... from...语句 1、查询array数组中的元素,使用索引来查找,比如: select name,subordinates[0] from employees; 2、查询map中的元素用 [键名] select name,deductions["State Taxes"] from employees; 3、查询struct中的元素用“点”符号 select name,address.city from employees; limit语句 限制返回的行数 select * from employees limit 2; case...when...then 句式 case...when...then语句和if条件语句类似,用于处理单个列的查询结果。例如: select name,salary, case when salary<5000.0 then 'low' when salary>=5000.0 and salary<7000.0 then 'middle' when salary>7000.0 then 'high' end as bracket from employees; 6.2 where语句 1、like和rlike like和rlike都可以进行模糊匹配,rlike是like的增强版,可以通过正则表达式来指定匹配条件,例如: select name,address.street from employees where address.street rlike '.*(Chicago|Ontario).*'; 6.3 group by 语句 having语句 having语句用来对group by产生的分组进行条件过滤,例如: 6.4、join语句 1、 内连接:只有进行连接的两个表中都存在与连接标准相匹配的数据才会被保留下来。Hive的join语句只支持等值连接,不支持“<” 和“>”。 select a.ymd, a.price_close, b.price_close from stocks a join stocks b on a.ymd=b.ymd where a.symbol='APPL' and b.symbol='IBM'; 2、 join语句优化:hive在执行join语句时,会把其他表都缓存起来,然后扫描最后那个表进行计算。所以在保证join语句中表从左至右要依次递增。比如: select a.name, b.companyname from company a join employees b on a.companycode=b.companycode,要保证表大小company <employees 。 3、 left outer join左外连接:左边表符合where条件的全部保留,对应右边表选择的列如果没有符合连接条件的将会返回null值。 4、 right outer join右外连接:返回右边表所有符合where语句的记录。左表中匹配不上的字段用null代替。 6.5 order by 、sort by、distribute by、cluster by 1、 order by:进行全局排序,会有一个所有的数据都通过一个reducer进行处理的过程。 2、sort by:进行局部排序,只会在每个reducer中对数据进行排序。 distribute by:distribute by和sort by配合使用。distribute by可以让具有相同字段值的记录被分发到一个reducer,实现了对数据进行分组,并根据sort by指定的字段进行组内排序。例如: select s.ymd, s.symbol, s.price_close from stocks s distribute by s.symbol sort by s.symbol ASC, s.ymd ASC;3、 cluster by:相当于distribute by...sort by...语句中distribute by 和sort by的是同一个字段,表示对数据进行分组显示。 select s.ymd, s.symbol, s.price_close from stocks s cluster by s.symbol; 七、HiveQL:视图 视图可以允许保存一个查询并像对待表一样对这个查询进行操作。这是一个逻辑结构,因为它不像一个表会存储数据。换句话说,Hive目前暂不支持物化视图。 使用视图的好处: 1、降低查询复杂度,例如下面以个嵌套子查询: from ( select * from people join cart on (cart.people_id=people.id) where firstname='john' ) a select a.lastname where a.id=3; 将嵌套子查询声明为一个视图: create view shorter_join as select * from people join cart on (cart.people_id=people.id) where firstname='john'; 这样就可以用这个视图来简化第一个查询: select lastname from shorter_join where id=3; 2、视图可以只展示一张表的部分内容,而不会让用户看到全部。 八、HiveQL:索引 Hive只有有限的索引功能,因为Hive没有主键和外键。建立索引可以加速查询速度。一张表的索引数据存储在另一张表中。 8.1 创建索引 例如对表employees的country字段建立索引: create index employees_index on table employees(country) as 'org.apache.hadoop.hive.ql.index.compact.CompactIndexHandler' with deferred rebuild idxproperties('creator'='me', 'created_at'='some_time') in table employees_index_table partitioned by (country,name) comment 'Employees indexed by country and name.'; employees_index表示索引名称,as...语句指定了索引处理器,也就是一个实现了索引接口的类,也可以使用其他的类(Hive内置的其他实现类、第三方实现类或者是自定义实现类),with deferred rebuild表示重建索引,这样新建的索引将呈现空白状态。in table...指定了一张新表,用来存储该索引数据。 8.2 Bitmap索引 bitmap索引应用于排重后值较少的列,如性别、年级等。 九、模式设计 9.1 按天划分的表 每天一张表的方式在数据库领域用来应对数据集增长很快的情况,通常会在表名中加入一个时间戳,比如upply_2011_01_01、upply_2011_01_02等等。对于hive,这种情况应该使用分区表,为每一天创建一个对应的分区。 create table supply(id int, part string, quantity int) partitioned by (int day); alter table supply add partition (day=20110102); alter table supply add partition (day=20110103); alter table supply add partition (day=20110104); 9.2 分区创建策略 一个理想的分区方案不应该导致产生太多的分区和文件夹目录,并且每个目录下的文件夹应该足够的大,应该是文件系统HDFS中块大小的若干倍。 9.3 同一份数据多种处理 hive提供了一个独特的语法,它可以从一个数据源产生多个数据聚合,而无需每次聚合都要重新扫描一次。例如,下面2个查询都会从源表history表读取数据,然后导入到2个不同的表中: insert overwrite table sales select * from history where action='purchased'; insert overwrite table credits select * from history where action='returned'; 上面的方式效率低下,下面这个查询可以达到同样的目的,却只需要扫描history表一次就可以: from history insert overwrite sales select * where action='purchased' insert overwrite credits select * where action='returned'; 9.4 分桶表存储数据 分桶是根据某一列的值把数据分成指定个数的文件进行存储,列值相同或者列值的hash值相同的记录会被存储到同一个文件。 分桶可以用于数据取样,同时当查询条件是分桶字段的话,也可以提高查询效率。 9.5 使用列存储表 hive通常使用行式存储,但是也提供了混合列式存储。下面两种表适合采用列式存储: (1)某些列的值会有很多重复的值,比如包含性别、年龄、状态等属性的表。 (2)表含有非常多的字段。 9.6 使用压缩 压缩都可以使磁盘上存储的数据量变小,这样可以通过降低I/O来提高查询执行速度。压缩和解压缩会消耗CPU资源,但是MapReduce任务通常是I/O密集型,所以几乎在所有场景下都要使用压缩。除了CPU密集型的场景,例如一些机器学习算法等。 十、调优 10.1 explain [extended] 使用explain关键字可以显示sql的执行计划,explain extended可以显示出更多的信息。 10.2 limit限制调整 一般情况下,Limit语句还是需要执行整个查询语句,然后再返回部分结果。这种情况通常是浪费的,应该尽可能的避免出现这种情况。 有一个配置属性可以开启,避免这种情况---对数据源进行抽样 hive.limit.optimize.enable=true --- 开启对数据源进行采样的功能 hive.limit.row.max.size --- 设置最小的采样容量 hive.limit.optimize.limit.file --- 设置最大的采样样本数 缺点:有可能部分数据永远不会被处理到。 10.3 本地模式 有时hive的输入数据量是非常小的。在这种情况下,为触发其他机器执行任务的时间消耗可能会比实际job的执行时间要多的多。对于大多数这种情况,hive可以通过本地模式在单台机器上处理所有的任务。对于小数据集,执行时间会明显被缩短 set hive.exec.mode.local.auto=true; 当一个job满足如下条件才能真正使用本地模式: 1.job的输入数据大小必须小于参数:hive.exec.mode.local.auto.inputbytes.max(默认128MB) 2.job的map数必须小于参数:hive.exec.mode.local.auto.tasks.max(默认4) 3.job的reduce数必须为0或者1 可用参数hive.mapred.local.mem(默认0)控制child jvm使用的最大内存数。 10.4 并行执行 hive会将一个查询转化为一个或多个阶段,包括:MapReduce阶段、抽样阶段、合并阶段、limit阶段等。默认情况下,一次只执行一个阶段。 不过,如果某些阶段不是互相依赖,是可以并行执行的。 set hive.exec.parallel=true,可以开启并发执行。 set hive.exec.parallel.thread.number=16; //同一个sql允许最大并行度,默认为8。会比较耗系统资源。 10.5 严格模式 Hive提供了一个严格模式,可以防止用户执行那些可能产生意想不到的不好的影响的查询。 通过设置hive.mapred.mode为strict来开启。 使用严格模式可以禁止3种类型的查询: (1)对于分区表,不加分区字段过滤条件,不能执行 (2)对于order by语句,必须使用limit语句。 (3)限制笛卡尔积的查询(join的时候不使用on,而使用where的)。 10.6 调整mapper和reducer数量 10.7 JVM重用 JVM启动过程比较慢,默认情况下,hadoop是通过为每一个map和reduce任务启动一个JVM实例来运行,当小文件比较多或者Task特别多的场景,任务的执行比较短,大量的时间消耗在JVM启动上。通过mapred.job.reuse.jvm.num.tasks配置,可以设置JVM实例在同一个job中重新使用的次数。这个功能的一个缺点是,开启JVM重用将会一直占用使用到的task插槽,直到整个任务执行结束。 10.8 索引 添加适当的索引来加快查询速度。 10.9 动态分区调整 (1)将动态分区模式设置为严格模式,必须保证至少有一个分区是静态的。通过设置hive.exec.dynamic.partition.mode为strict。 (2)限制单表可以创建的最大分区数。hive.exec.max.dynamic.partitions。 十二、函数 12.1 内置函数 12.1.1 标准函数 以一行数据中的一列或多列数据作为参数然后返回结果是一个值的函数,大多数函数都是属于这类的,比如round()、floor()这些数学函数。 12.1.2 聚合函数 聚合函数接受从零行到多行的零个到多个列,然后返回单一值。例如sum()、avg()、min()、max()等。 12.1.3 表生成函数 表生成函数接受零个或多个输入,然后产生多列或多行输出,例如explode()函数以array类型数据作为输入,然后对数组中的数据进行迭代,返回多行结果,一行一个数组元素值。 12.2 自定义函数 分别提供了三种接口对应标准函数、聚合函数、表生成函数,需要实现对应接口,实现接口的方法。 十三、Streaming 自定义函数需要开发人员编写java代码,对于不熟悉java的开发人员,hive提供了streaming的方式来完成同样的事情。streaming就是在hiveQL中允许使用系统的脚本来对数据进行处理。 例如,使用Linux系统中的命令cat来查询表: select transform(e.name, e.salary) using '/bin/cat' as name, salary from employee e; 同时还允许运行用户自定义的脚本。 十四、文件格式、记录格式和文件压缩 14.1 文件格式 Hive内置了三种文件格式:文本文件、SequenceFile、RCfile。当从表中读取数据时,Hive会使用到InputFormat,向表中写入数据时,会使用OutputFormat。在create table...建表时,通过stored as...来指定文件格式。 14.1.1 文本文件 文本文件格式是默认存储格式,等价于在创建表时通过 stored as textfile语句指定使用文本存储格式。对应的InputFormat是TextInputFormat,OutputFormat是HiveIgnoreKeyTextOutputFormat。 14.1.2 SequenceFile SequenceFile是hadoop本身就支持的一种标准文件格式,在定义表结构时通过stored as sequencefile语句指定。对应的InputFormat是SequenceFileInputFormat,OutputFormat是HiveSequenceFileOutputFormat。 14.1.3 RCfile 采用列式存储。一张表有成百上千字段,而大多数的查询只需要使用到其中的一小部分字段,这时候该表就适合采用RCfile进行存储。对应的InputFormat是RCFileInputFormat,OutputFormat是RCFileOutputFormat。 14.1.4 自定义文件格式 可以通过实现InputFormat和OutFormat来自定义文件格式。在定义表结构时,可以指定InputFormat和OutFormat,例如: create table colunmTable(key int, value int) row format serde 'org.apache.hive.serde2.columnar.ColumnarSerde' stored as inputformat 'org.apache.hadoop.hive.ql.io.RCFileInputFormat' outputformat 'org.apache.hadoop.hive.ql.io.RCFileOutputFormat' 14.2 记录格式 14.2.1 SerDe SerDe是序列化/反序列化的接口,在内部,Hive引擎使用定义的InputFormat来读取一行数据记录,这行记录之后会被传递给SerDe.deserialize()方法来处理,SerDe决定了一行记录如何被解析。Hive内置了几个SerDe,比如处理正则表达式RegexSerDe,CSV文件CSVSerDe。 14.2.2 JSON SerDe JsonSerde来自于第三方,可以用于查询json格式的记录。例如: create external table messages ( msg_id bigint, tstamp string, text string, user_id bigint, user_name, string ) row format serde "org.apache.hadoop.hive.contrib.serde2.JsonSerde" with serdeproperties ( "msg_id"="$.id", "tstamp"="$.created_at", "text"="$.text", "user_id"="$.user.id", "user_name"="$.user.name" ) location '/data/messages'; row format serde用于指定使用的SerDe,with serdeproperties用于给SerDe传递属性信息,上面的例子中用这些属性将json文档和表的字段对应起来。定义好之后,就可以像通常一样执行查询:select * frommessages; 14.3 文件压缩 压缩可以节省磁盘存储空间,减小磁盘和网络I/O时间。Hive表的文件存储在hadoop上,可以使用hadoop提供的压缩对文件进行压缩。除此以外,hive也提供了两个跟压缩相关的特有设置: (1)开启中间压缩 hive.exec.compress.intermediate 对中间数据进行压缩可以减少job中map和reduce task间的数据传输量 (2)最终输出结果压缩 hive.exec.compress.output和mapred.output.compression.codec 十五、存储处理程序和NoSQL 存储处理程序可以将外部实体表作为标准的hive表进行处理,它包含了InputFormat、OutputFormat、SerDe和Hive需要使用的特定的代码。例如可以将Hbase、Cassandra、DynamoDB这样的NoSQL数据库中的表作为hive表来处理,下面是创建一个指向Hbase表的Hive表: create table habase_stocks(key int, name string, price float) stored by 'org.apache.hive.hbase.HBaseStorageHandler' with serdeproperties("hbase.columns.mapping"=":key,stock:val") tblproperties("hbase.table.name"="stocks"); 十六、权限 默认情况下,授权模块是不开启的,需要将属性hive.security.authorization.enabled设置为true。 16.1 用户、组和角色 hive中的用户和组是外部系统的用户和组,比如当使用HiveCLI方式来访问hive时,此时的用户和组就对应linux系统当前用户和组。角色是hive自己创建的,它的使用比较灵活。可以为一个角色赋予权限,然后给用户设置该角色,这样用户就拥有了该角色所拥有的权限。 16.2 权限种类 下面列举了可以配置的权限: 名称 描述 all 所有的权限 alter 修改表结构的权限 create 创建表的权限 drop 删除表或表中的分区的权限 index 创建表索引的权限 lock 开启并发后,锁定和解锁表的权限 select 查询表或者分区中数据的权限 show_database 查看所有数据库的权限 update 向表或者分区中插入或加载数据的权限 例如为用户edward赋予在数据库edsstuff中create权限: grant create on database edsstuff to user edward; 16.3 分区级别的授权 默认情况下,是在表级别授予权限的,但是可以在分区级别进行授权,将表属性partition_level_privilege设置为true即可。如下面的例子: create table autorized_part (key int, value string) partitioned by (ds string); //创建一张分区表autorized_part alter table autorized_part set tblproperties("partition_level_privilege"="true");//为该表开启分区级别的授权 alter table autorized_part add partition (ds='3'); //为表新增一个分区,值为3 revoke select on table autorized_part partition(ds='3') from user edward; //给edward赋予分区ds='3'的select 权限 十七、锁 hive不支持行级别的更新和删除,也不支持事务,因此细粒度锁对hive是没有必要的。hive提供了表级别和分区级别的锁,这需要配合zookeeper来使用。 在hive-site.xml配置文件中,增加hive.zookeeper.quorum属性来指定zookeeper集群的地址,同时设置hive.support.concurrency为true,开启支持并发。 show locks; //查询当前所有锁 show locks 表名; //查询某张表上的锁 show locks 表名 partition (...); //查询某个分区上的锁 17.1 锁的种类 hive中的锁分为两种:共享锁和独占锁。共享锁和共享锁之间不互斥,通常在读取的时候使用共享锁,独占锁通常在写入的时候使用,它会使其他用户无法访问锁定表或分区。

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

Python笔记

一、基础 None:空值 0x :16进制前缀; 0 :8进制前缀 x, y = y, x:交换值 #:单行注释 False = 0 = [] = {} 运算符 +:可用于数字运算、序列连接、字符串连接 *:可用于数字运算、序列重复、字符串重复 **:幂运算,优先级最高的符号运算符 in:检测是否存在其中,可用于检测字符串、序列、 输出 print("My name is Li."):打印输出 print(r'etc\drivers\hosts'):打印原始字符串(不转义) print(u'1234567'):打印 Unicode 字符串(3.0后,字符串全都是 Unicode 字符串) 输入 input("Your name: "):获取输入 raw_input("Your age: ") 将接收值转换为字符(常用)(3.0时被重命名为 input) 导入 import math:导入库(模块) from math import sqrt as sqrt1:导入函数并取别名 cmath 为 complete math,其函数可处理虚数等复杂运算 二、通用列表 索引取值 numbers[-1]: 倒数第一个元素 numbers[3:6]:取子集 【得到(6 - 3)个数】 numbers[-2:]:取最后 2个数的子集 numbers[:3]:取前面 3个数的子集 numbers[:]:复制整个序列 numbers[0:10:2]:按 歩长 取数的子集 【分别取得第1,3,5,7,9个数】 numbers[0:10:-2]: 反方向 取得子集 【分别取得第10,8,6,4,2个数】 基本操作 numbers[2:2] = list('hello'):插入 5个元素 numbers[3:] = list('world'):替换第 4个元素及之后的元素 numbers[1:4] = []:删除某些值,效果同 del 基本方法 append(9):追加单个元素 count('1'):统计某个元素的出现次数 extend(arr2):在 原列表 中追加另一个列表的元素 index('hi'):凭值找出元素的位置,不存在则引发异常 insert(3, 'app'):在指定位置插入元素 pop():移除元素,默认最后一个 remove('app'):移除指定值的第一个匹配的元素 reverse():将列表反向存放,无返回值 sort():排序原列表,可带参(函数)表示排序方式 sorted(arr):返回已排序的列表副本 元组 值不可变 用逗号分隔 tuple 将序列(列表或字符串等)转换为元组 常用于映射和返回值 三、字符串 格式化 format = "Hello, we have %s%% discount. Only $$%s." values = ('90', 18) # 自动将 18 转换为字符串 print format % values 具体类型转换格式: d, i 十进制整数 o 不带符号的八进制 u 不带符号的十进制 x, X 不带符号的十六进制 e, E 浮点数 f, F 十进制浮点数 g, G 浮点数,精度值等于 e (指数大于 -4)或 f C 单字符 r 字符串(使用 repr 转换) s 字符串(使用 str 转换) 转换标志 - 左对齐 + 加符号 0 补够位数 '' 空格,正数前保留空格 方法 find('child'):找子串,返回所在开始位置,找不到则返回 -1 '\'.join(strs):用'斜杠'连接序列(元素全是字符串) lower():返回小写版 replace(a, b):在字符串中用串 b 替换子串 a split(','):用'逗号'分隔字符串,存放于序列中 strip():去除两侧的空格 ...... 四、字典 创建和使用 dict = {'Mon' : 1, 'Tue': 2, 'Web': 3} dict['Tue'] 方法 clear():清空原始字典的项 copy():返回副本(浅复制) {}.fromkeys(['name', 'age'], 0):使用键建立字典,值全为 0(第二参数不填则默认为 None) get('age'):取键值,不存在则返回 None has_key('sex'):检测是否存在键 items():将字典以列表方式,每项转换为(键,值)形式 keys():将键转换为列表 pop('name'):返回指定值,删除指定键与值 popitem():随机弹出值,并删除键 setdefault('name', 100):获取值,若键不存在则设定键,值为 100(无第二参数则默认值为 None) update(dict2):利用一个字典更新原字典 values():将值转换为列表 五、常用函数 min() max() len() list('hi'):返回序列 ['h', 'i'] cmp(a, b):比较两个数,返回 1,0,-1 del a删除变量 a的值(移除引用) exec 'print("Hello world!")':执行语句 eval():求值,同 input() pass:什么也不做,充当占位符 条件、循环 if name == 'John': print(name) elif(name == 'Boss'): print("My lord.") else: print("None.") while i < 100: i += 1 for num in numbers: print(str(num) + '\n') 其他比较运算符 x is y:是否为同一对象(同一引用) x is not y x in y:成员资格运算符 x not in y 迭代工具 names = ['a', 'b', 'c', 'd'] ages = [12, 13, 11, 14] for i in range(len(names)): print(names[i], 'is' ages[i], ' years old.') # 逗号等价于 + ' ' + for name, age in zip(names, ages): print(name, 'is' age, ' years old.') 六、抽象 函数 支持默认值 *param:单星号,不定量参数 **dict:双星号,传递元组或字典 类 __metaclass__ = type class Person: def setName(self, name): self.name = name def getName(self): return self.name def greet(): print("Hello , I'm %s." % self.name) man = Person() man.greet() # 等价于 Person.greet(man)

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

UnitTest笔记

UnitTest——测试模块 其他相关:使用coverage工具统计python单元测试覆盖率 一、简单示例 待测类/方法 book.py # 待测类 class Book(object): _page = 2000 def __init__(self, bookname, author): self.name = bookname self.author = author def get_page(self): return self._page # 待测方法 def get_name(book): return book.name 测试类test_book.py # 每个测试方法均以 test 开头,否则是不被unittest识别的。 class TestBook(unittest.TestCase): def test_init(self): print('test init()') book = main_class.Book('bookname', 'author') self.assertEquals(book.name, 'bookname') self.assertEquals(book.author, 'author') self.assertEquals(book._page, 2000) self.assertTrue(isinstance(book, main_class.Book)) def test_get(self): print('test get_name()') self.assertEquals(main_class.get_name(main_class.Book('bookname', 'author')), 'bookname') 运行 if __name__ == '__main__': unittest.main() 结果 test get_name() . test init() . ---------------------------------------------------------------------- Ran 2 tests in 0.001s OK ​ 给出的结果标识,成功是 .,失败是 F,出错是 E,跳过是 S。 ​ 测试的执行跟方法的顺序没有关系。 二、测试类 创建 继承unittest.TestCase。 以“test”开头的测试方法,每一类分别添加一个测试方法。 按需添加内置判断条件如:assertEquals()、assertRaises(Error)。 运行 1、pyCharm直接运行 ​ pycharm还是比较智能的,可以直接运行测试,不过有个地方需要特别注意,pyCharm会自动识别并测试鼠标所在当前代码块,所以有可能你ctrl+shift+f10后,可能运行结果就只是测试一个方法(笔者可是让这个弄的莫名其妙,一顿不得其解)。 2、main方法 ​ 在测试文件结尾添加 if __name__ == '__main__': unittest.main() 然后命令行运行 python xxxx.py 3、命令行运行 在命令行通过参数-m unittest直接运行单元测试: $ python -m unittest xxxx 这是推荐的做法,因为这样可以一次批量运行很多单元测试,并且,有很多工具可以自动来运行这些单元测试。 环境准备和尾处理 如果每个测试之前都需要准备环境如连接数据库、打开文件,或尾处理如关闭数据库、关闭数据流,不可能每个测试方法都写一遍。 setUp() ​ 在每个测试方法运行之前运行 tearDown() ​ 在每个测试方法运行之后运行 setUpClass() ​ 在每个单元测试开始之前运行 tearDownClass() ​ 在每个单元测试结束之后运行 示例: class TestBook(unittest.TestCase): def setUp(self): print('setUp++++++++') @classmethod def setUpClass(cls): print('setUpClass--------') def tearDown(self): print('tearDown++++++++++') @classmethod def tearDownClass(cls): print('tearDownClass--------') def test_init(self): print('test init()') def test_get(self): print('test get_name()') 运行结果: (UnitTest) D:\sHui\learn\UnitTest>python -m unittest test_class setUpClass-------- setUp++++++++ test get_name() tearDown++++++++++ .setUp++++++++ test init() tearDown++++++++++ .tearDownClass-------- ---------------------------------------------------------------------- Ran 2 tests in 0.003s OK 跳过某case 测试跳过某个case?unittest提供了几种方法。 1、skip装饰器 unittest.skip(reason) skip无条件跳过 unittest.skipIf(condition, reason) skipIf当condition为True时跳过 unittest.skipUnless(condition, reason) skipUnless当condition为False时跳过。 示例: class TestBook(unittest.TestCase): def test_init(self): print('test init()') @unittest.skip('skip this func ') def test_get(self): print('test get_name()') 结果: Skipped: skip this func test init() Ran 2 tests in 0.002s OK (skipped=1) 运行了两个测试,但test_get被跳过了 2、TestCase.skipTest() 测试类的skipTest()方法可以跳过当前测试方法。 class TestBook(unittest.TestCase): def test_init(self): print('test init()') # 跳过此方法 def test_get(self): self.skipTest('skip this func') print('test get_name()') 结果同上 调整测试顺序 新建文件test_suite.py import unittest import test_class if __name__ == '__main__': suite = unittest.TestSuite() tests = [test_class.TestBook('test_init'), test_class.TestBook('test_get')] suite.addTests(tests) runner = unittest.TextTestRunner(verbosity=2) runner.run(suite) 输出: test init() test get_name() test_init (test_class.TestBook) ... ok test_get (test_class.TestBook) ... ok ---------------------------------------------------------------- Ran 2 tests in 0.000s OK 三、测试输出 1、输出信息 verbosity参数可以控制执行结果的输出 0 是简单报告 1 是一般报告 2 是详细报告。 2、输出到文件 if __name__ == '__main__': suite = unittest.TestSuite() suite.addTests(unittest.TestLoader().loadTestsFromTestCase(test_class.TestBook)) with open('Unittest.txt', 'a') as f: runner = unittest.TextTestRunner(stream=f, verbosity=2) runner.run(suite) 运行结束会在同目录下生成了Unittest.txt 3、输出为HTML HTMLTestRunner是一个第三方的unittest HTML报告库,首先我们下载HTMLTestRunner.py,并放到当前目录下,或者你的’C:Python27Lib’下,就可以导入运行了。 官方原版:下载地址 灰蓝修改版:HTMLTestRunner.py(已调整格式,中文显示) 示例: from HTMLTestRunner import HTMLTestRunner if __name__ == '__main__': suite = unittest.TestSuite() suite.addTests(unittest.TestLoader().loadTestsFromTestCase(test_class.TestBook)) with open('HTMLReport.html', 'w') as f: runner = HTMLTestRunner(stream=f, title='MathFunc Test Report', description='generated by HTMLTestRunner.', verbosity=2) runner.run(suite) 四、Coverage的使用 1.命令行方式 详见:http://coverage.readthedocs.io/en/latest/cmd.html 可以使用help命令查看帮助:$ coverage help 关键命令如下: A.run 执行代码覆盖率统计,只需要通过coverage的run参数执行被统计代码即可。 $ coverage run test.py arg1 arg2 test.py是测试脚本,arg1 arg2是test.py执行需要的参数。跑完后,会自动生成一个覆盖率统计结果文件(data file):.coverage。 B.report 有了覆盖率统计结果文件,只需要再运行report参数,就可以在命令里看到统计的结果。 Stmts/Miss表示语句总数/未执行到的语句数 Cover=(Stmts-Miss)/Stmts c. html 生成html的测试报告。 $ coverage html -d covhtml 生成的报告直接关联代码,高亮显示覆盖和未覆盖的代码,支持排序。-d指定html文件夹。 2.API方式 除了使用命令行,还可以在python代码中直接调用coverage模块执行代码覆盖率的统计。使用方法也非常简单: import coverage cov = coverage.coverage(source = ['totest']) cov.start() #coding cov.stop() cov.report() cov.html_report(directory='covhtml') source指定要执行统计的文件,source = ['totest']只统计totest.py的覆盖率 directory指定生成html的路径 五、常用函数 Method Checks that assertEqual(a, b) a == b assertNotEqual(a, b) a != b assertTrue(x) bool(x) is True assertFalse(x) bool(x) is False assertIs(a, b) a is b assertIsNot(a, b) a is not b assertIsNone(x) x is None assertIsNotNone(x) x is not None assertIn(a, b) a in b assertNotIn(a, b) a not in b assertIsInstance(a, b) isinstance(a, b) assertNotIsInstance(a, b) not isinstance(a, b) assertRaises(exc, fun, *args, **kwds) fun(*args, **kwds) raises exc assertRaisesRegexp(exc, r, fun, *args, **kwds) fun(*args, **kwds) raises exc and the message matches regex r assertAlmostEqual(a, b) round(a-b, 7) == 0 assertNotAlmostEqual(a, b) round(a-b, 7) != 0 assertGreater(a, b) a > b assertGreaterEqual(a, b) a >= b assertLess(a, b) a < b assertLessEqual(a, b) a <= b assertRegexpMatches(s, r) r.search(s) assertNotRegexpMatches(s, r) not r.search(s) assertItemsEqual(a, b) sorted(a) == sorted(b) and works with unhashable objs assertDictContainsSubset(a, b) all the key/value pairs in a exist in b assertMultiLineEqual(a, b) strings assertSequenceEqual(a, b) sequences assertListEqual(a, b) lists assertTupleEqual(a, b) tuples 参考文章: 官方文档 Python必会的单元测试框架 —— unittest 廖雪峰

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

js笔记

1.window.open(''url'') 2.用自定义函数 <script> functionopenWin(tag,obj) { obj.target="_blank"; obj.href="Web/Substation/Substation.aspx?stationno="+tag; obj.click(); } </script> <ahref="javascript:void(0)"onclick="openWin(3,this)">株洲</a> 3、window.location.href="" checkbox选中 4、有以下一组checkbox列,在修改页面上如何根据从数据库里获取的数据来设置这里的checkbox选中与否: 这个提交到数据库后数据格式是这样子的:1, 2, 3, 5, 8, 9, 12 假设此字段为strB,假设checkbox的name="A" 下边的js脚本就是实现默认选中的: <script language="JavaScript"> <!-- var B="<%=strB%>"; with(document.form1) { for (i=0;i<A.length ;i ) { tmpB=B.split(", "); for (j=0;j<tmpB.length ;j ) { if(tmpB[j]==A[i].value) {A[i].checked=true;break;} } } } //--> </script> 本文转自 00_yatou 51CTO博客,原文链接:http://blog.51cto.com/ql0722/1656011,如需转载请自行联系原作者

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

hue笔记

创建Hue用户 建议为hue创建专职的用户,同时将/usr/local/hue的group和owner设给hue用户: groupadd hue useradd -g hue hue chown -R hue:hue /usr/local/hue 1 2 3 启动 启动: sudo -u hue nohup /usr/local/hue/build/env/bin/supervisor & 1 停止: pkill -U hue 1 常见问题 HUE页面提示:The app won’t work without a running Livy Spark Server Livy Server是Spark的Rest API服务,官方文档: https://github.com/cloudera/hue/tree/master/apps/spark/java 我们可以使用: sudo -u hue nohup /usr/local/hue/build/env/bin/hue livy_server & 1 来启动它,但是在启动前你需要了解它对Spark的版本依赖。Hue3.9中的Livy Server需要Spark 1.4,而HDP2.3安装的是Spark 1.3,也就是说如果你安装是的HDP,现在还不能在Hue上使用Spark,为了避免HUE页面上报:Spark The app won’t work without a running Livy Spark Server的错误提示,我们可以禁用Spark模块,同样是编辑hue.ini文件: app_blacklist=impala,security,spark 1 2 HUE页面报错:/wsgiserver.py line 1196 in communicatereq.respond() 如果打开HUE页面时出现: Traceback (most recent call last): File “/usr/local/hue/desktop/core/src/desktop/lib/wsgiserver.py”, line 1196, in communicate req.respond() 很有可能是早先使用rpm安装旧版本的Hue之后没有完全卸载干净的原因,除了检查相关文件是否已经彻底删除之外,务必删除早先建立的hue用户和用户组。 本文转自 yntmdr 51CTO博客,原文链接:http://blog.51cto.com/yntmdr/1747148,如需转载请自行联系原作者

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

hadoop笔记

一、hadop的项目 Common:一系列组件和接品口,用于分布式文件系统和通用I/O(序列化,Java RPC 和持久化数据结构) Avro:一种序列化系统,用于支技高效,跨语言的RPC和持久化的数据存储 MapRedue:分布式数据处理模型和执行环境,运行于大型商用机集群 HDFS:分布式文件系统,运行于大型商用机集群 Pig:数据流语言和运行环境,用以探究非常庞大的的数据集。Pig运行在MapReduce和HDFS集群上 Hive:一种分布式,按列存储的数据仓库。Hive管理HDFS中存储数据,并提供基于SQL的查询语言(由运行时引擎翻译成MapReduce作业)用以查询数据 HBase:一种分布式,按列存储的数据库。HBase使用HDFS作为底层存储,同时支MapReduce的批量式计算和点查询(随机读取) Zookeeper:一种分布式,可用性高的协调服务。Zeekeeper提供分布式锁之类的基本服务用于构建分布式应用。 Sqooq:该工具用于在结构化数据存储(如关系型数据库)和HDFS之间高效批量传输数据。 Oozie:该服务用于运行和调度Hadoop作业(如MapReduce,Pig,Hive及sqoop)。 二、hadoop分布式文件系统 1、HDFS设计 存府超大文件 流式数据访问(一次写入多次读取) 商用硬件(不需要豪华配置) 低时间延迟的数据访问 大量的小文件 多用户写入,任意修改文件 2、数据块 磁盘一般512字节 HDFS默认64M 与其他文件系统不同的是,HDFS中小于一块大小的文件不会占据整个块的空间 HDFS的块比磁盘的块大,主要为了最小化寻址开销 块抽象好处。(一,一个文件的大小可以大于网络任意一个磁盘的容量。因为文件的所有块并不需要存储在同一个磁盘上。二,使用抽象块而非整个文件作为存储单元,简化了存储子系统的设计。三,块适用于数据备份,提供数据容错能力和提高可用性) 三、namenode和datanode HDFS以管理者-工作者模式运行,一个NN和多个DN,NM管理文件系统的命名空间,维护着文件系统树及整棵树内的所有文件和目录。这些信息保存在(命名镜像文件)和(编辑日志文件);存储元数据与文件到数据块映射的地方 DN是文件系统工作节点,存储并检索数据块,定期向NN发送它们存储块列表。 NN很重要,如果运行NM服务器损块,文件系统上的所有文件将会丢失。所以对NM容错,提供两种机制。一,备份那些组成文件系统元数据持久状态文件。hadoop可以配置使NN在多个文件系统上保存元数据的持久状态。写操作实时同步,是原子操作。一般配置,将持久状态写入本地磁盘的同时,写入一个远程挂载的网络文件系统(NFS)。二; 运行一个辅助NN,但它不能被用作NN,这个辅助NN的重要作用是定期通过编辑日志合并命名空间镜像,以防止编辑日志过大。这个SNN一般在另一台单独的无力计算机上运行,以为他需要占用大量的cpu和NN相同容量的内存来执行合并操作。它会保存合并后的命名空间镜像的副本,并在NN发生故障时启用。但是SNN滞后与NN,所以如果NN全部失效,难免会丢失部分数据。在这种情况下,一般把存储在NFS上的NN元数据复制到SNN并作为新的NN运行。 四、网络拓扑与hadoop 将两个节点间带宽作为距离的衡量标准,以下场景,可以可用带宽依次递减 同一节点上进程 同一机架上的不同节点 同一数据中心不同机架上的节点 不同数据中心的节点 五、数据流 1、剖析文件读取 客户端调用FileSystem 对象的open()方法来打开希望读取的文件, 1.首先调用FileSystem对象的open方法,其实是一个DistributedFileSystem的实例 2.DistributedFileSystem通过rpc获得文件的第一批个block的locations,同一block按照重复数会返回多个locations,这些locations按照hadoop拓扑结构排序,距离客户端近的排在前面. 3.前两步会返回一个FSDataInputStream对象,该对象会被封装成DFSInputStream对象,DFSInputStream可以方便的管理datanode和namenode数据流。客户端调用read方法,DFSInputStream最会找出离客户端最近的datanode并连接(参考第一小节)。 4.数据从datanode源源不断的流向客户端。 5.如果第一块的数据读完了,就会关闭指向第一块的datanode连接,接着读取下一块。这些操作对客户端来说是透明的,客户端的角度看来只是读一个持续不断的流。 6.如果第一批block都读完了,DFSInputStream就会去namenode拿下一批blocks的location,然后继续读,如果所有的块都读完,这时就会关闭掉所有的流。 如果在读数据的时候,DFSInputStream和datanode的通讯发生异常,就会尝试正在读的block的排第二近的datanode,并且会记录哪个datanode发生错误,剩余的blocks读的时候就会直接跳过该datanode。DFSInputStream也会检查block数据校验和,如果发现一个坏的block,就会先报告到namenode节点,然后DFSInputStream在其他的datanode上读该block的镜像 该设计的方向就是客户端直接连接datanode来检索数据并且namenode来负责为每一个block提供最优的datanode,namenode仅仅处理block location的请求,这些信息都加载在namenode的内存中,hdfs通过datanode集群可以承受大量客户端的并发访问。 2、剖析文件写入 客户端通过调用DistributedFileSystem的create方法创建新文件 DistributedFileSystem通过RPC调用namenode去创建一个没有blocks关联的新文件,创建前,namenode会做各种校验,比如文件是否存在,客户端有无权限去创建等。如果校验通过,namenode就会记录下新文件,否则就会抛出IO异常. 前两步结束后会返回FSDataOutputStream的对象,象读文件的时候相似,FSDataOutputStream被封装成DFSOutputStream.DFSOutputStream可以协调namenode和datanode。客户端开始写数据到DFSOutputStream,DFSOutputStream会把数据切成一个个小packet,然后排成队列data quene。 DataStreamer会去处理接受data quene,他先问询namenode这个新的block最适合存储的在哪几个datanode里(参考第二小节),比如重复数是3,那么就找到3个最适合的datanode,把他们排成一个pipeline.DataStreamer把packet按队列输出到管道的第一个datanode中,第一个datanode又把packet输出到第二个datanode中,以此类推。 DFSOutputStream还有一个对列叫ack quene,也是有packet组成,等待datanode的收到响应,当pipeline中的所有datanode都表示已经收到的时候,这时akc quene才会把对应的packet包移除掉。 如果在写的过程中某个datanode发生错误,会采取以下几步:1) pipeline被关闭掉;2)为了防止防止丢包ack quene里的packet会同步到data quene里;3)把产生错误的datanode上当前在写但未完成的block删掉;4)block剩下的部分被写到剩下的两个正常的datanode中;5)namenode找到另外的datanode去创建这个块的复制。当然,这些操作对客户端来说是无感知的。 客户端完成写数据后调用close方法关闭写入流 DataStreamer把剩余得包都刷到pipeline里然后等待ack信息,收到最后一个ack后,通知datanode把文件标示为已完成。 六、hadoop的I/O操作 1、HDFS的数据完整性 HDFS对写入的所有数据计算校验和,并在读取数据时验证交验和,它针对每个由io.bytes.per.checksum指定字节的数据计算校验和。默认512字节,由于CRC-32校验和是4个字节,所以存储校验和的额外开销低于1% 客户端在读取数据块时,如果检测到错误,首先向namenode报告已损坏的数据块及其正在尝试读操作的这个datanode,再抛出ChecksumException异常。namenode将这个数据块复本标记为已损坏,因此,它将不会处理请求直接发送到这个节点,或尝试将这个复本复制到另一个datanode。之后,安排这个数据块的一个复本复制到另一个datanode。 2、序列化 序列化是指将结构化对象转化为字节流以便在网络上传输的或写到磁盘进行永久存储过程。反序列是指将字节流转回结果化的逆向过程。 序列化在分布式数据处理的两大领域常出现:进程间通信和永久存储。 hadoop中,系统中多个节点上进程间通信是通过“远程过程调用(remote procedure call,RPC)实现”,RPC序列化格式如下 紧凑 紧凑格式能充分利用网络带宽(数据中心最稀缺的资源) 快速 进程间通信形成了分布式系统的骨架,尽量减少序列化和反序列的性能开销 可扩展 满足新需求,协议不断变化 支持互操作 支技以不同语言写的客户端与服务器交互 hadoop使用自己序列化格式writable,紧凑,速度化。但不太容易用java以外语言进行扩燕尾服或使用。Writable是hadoop核心(大多数的MapReduce程序都为会键和值使用它) 本文转自 zouqingyun 51CTO博客,原文链接:http://blog.51cto.com/zouqingyun/1710619,如需转载请自行联系原作者

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

Docker笔记

简介 Docker 是 dotCloud 最近几个月刚宣布的开源引擎,旨在提供一种应用程序的自动化部署解决方案,简单的说就是,在 Linux 系统上迅速创建一个容器(类似虚拟机)并在容器上部署和运行应用程序,并通过配置文件可以轻松实现应用程序的自动化安装、部署和升级,非常方便。因为使用了容器,所以可以很方便的把生产环境和开发环境分开,互不影响,这是 docker 最普遍的一个玩法。更多的玩法还有大规模 web 应用、数据库部署、持续部署、集群、测试环境、面向服务的云计算、虚拟桌面 VDI 等等。 Docker 使用 Go 语言编写,用 cgroup 实现资源隔离,容器技术采用 LXC. LXC 已经足够成熟,被多个主流 PaaS 服务商采用(比如 dotCloud),国内的一些互联网公司也在用(比如腾讯)。虽然都是企图解决自动化部署方面的问题,Docker 的解决方式有别于我们常提到的 Puppet/Chef,他们虽然走的是不同的路,但也可以拿来一起用。 安装 (ubuntu 12.04) Due to a bug in LXC, Docker works best on the 3.8 kernel. Precise comes with a 3.2 kernel, so we need to upgrade it. The kernel you’ll install when following these steps comes with AUFS built in. We also include the generic headers to enable packages that depend on them, like ZFS and the VirtualBox guest additions. sudo apt-get install python-software-properties sudo apt-get install linux-image-generic-lts-raring linux-headers-generic-lts-raring sudo reboot 添加docker源并安装docker sudo apt-key adv --keyserver keyserver.ubuntu.com --recv-keys 36A1D7869245C8950F966E92D8576A8BA88D21E9 sudo sh -c "echo deb http://get.docker.io/ubuntu docker main > /etc/apt/sources.list.d/docker.list" sudo apt-get update sudo apt-get install lxc-docker 上面两步也可以替换成执行curl -s https://get.docker.io/ubuntu/ | sudo sh,更简单。 创建容器 首先从https://index.docker.io/下载一个预定义的镜像 sudo docker pull ubuntu 启动一个容器: sudo docker run -i -t ubuntu /bin/bash Starting a long-running worker process # Start a very useful long-running process JOB=$(sudo docker run -d ubuntu /bin/sh -c "while true; do echo Hello world; sleep 1; done") # Collect the output of the job so far sudo docker logs $JOB # Kill the job sudo docker kill $JOB Bind a service on a TCP port # Bind port 4444 of this container, and tell netcat to listen on it JOB=$(sudo docker run -d -p 4444 ubuntu:12.10 /bin/nc -l 4444) # Which public port is NATed to my container? PORT=$(sudo docker port $JOB 4444 | awk -F: '{ print $2 }') # Connect to the public port echo hello world | nc 127.0.0.1 $PORT # Verify that the network connection worked echo "Daemon received: $(sudo docker logs $JOB)" Committing (saving) a container state¶ Save your containers state to a container image, so the state can be re-used. When you commit your container only the differences between the image the container was created from and the current state of the container will be stored (as a diff). See which images you already have using the docker images command. # Commit your container to a new named image sudo docker commit <container_id> <some_name> # List your containers sudo docker images You now have a image state from which you can create new instances. 更多功能见http://docs.docker.io/ 什么是docker Docker 是 Docker.Inc 公司开源的一个基于LXC技术之上构建的Container容器引擎, 源代码托管在 GitHub 上, 基于Go语言并遵从Apache2.0协议开源。 Docker在2014年6月召开DockerConf 2014技术大会吸引了IBM、Google、RedHat等业界知名公司的关注和技术支持,无论是从 GitHub 上的代码活跃度,还是Redhat宣布在RHEL7中正式支持Docker, 都给业界一个信号,这是一项创新型的技术解决方案。 docker的基本概念 ①镜像:用来创建Docker容器的只读模板 ②容器:从镜像创建而来的运行实例,各个容器之间相互隔离 ③仓库:存放镜像的场所,如https://hub.docker.com/和http://www.dockerpool.com/ 安装docker 对于centos7,直接执行yum -y install docker即可安装,安装完成后需要执行systemctl start docker来启动docker服务。 其他操作系统中的安装方法见https://docs.docker.com/installation/#installation 获取镜像 从Docker Hub仓库下载一个Ubuntu 12.04操作系统的镜像 $ sudo docker pull ubuntu:12.04 Pulling repository ubuntu ab8e2728644c: Pulling dependent layers 511136ea3c5a: Download complete 5f0ffaa9455e: Download complete a300658979be: Download complete 904483ae0c30: Download complete ffdaafd1ca50: Download complete d047ae21eeaf: Download complete 官方镜像比较慢的时候可以从其他仓库下载镜像,如 $ sudo docker pull www.dockerpool.com:5000/library/centos:centos7 查询本地已下载的镜像 $ sudo docker images REPOSITORY TAG IMAGE ID CREATED VIRTUAL SIZE ubuntu 12.04 0a51fcc173d8 2 days ago 111 MB centos httpd 9ad57da2b81c 6 weeks ago 336.5 MB e01000c7bac8 6 weeks ago 336.5 MB centos latest b157b77b1a65 8 weeks ago 243.7 MB 其中,镜像id唯一标识了镜像,TAG信息用来标记来自同一个仓库的不同镜像。例如ubuntu仓库中有多个镜像,通过TAG信息来区分发行版本,例如10.04、12.04、12.10、13.04、14.04等。创建容器时,如果不指定具体的标记,则默认使用latest标记信息。 除了从容器中下载已有镜像外,也可以根据已有镜像创建新的镜像。创建新镜像有多种方法: ①修改已有镜像后commit $ sudo docker run -t -i ubuntu:12.04 /bin/bash root@a439b6e894bb:/# apt-get update root@a439b6e894bb:/# apt-get install nginx root@a439b6e894bb:/# /etc/init.d/nginx start root@a439b6e894bb:/# exit $ sudo docker commit -m 'add ngnix' -a 'feisky' a439b6e894bb ubuntu:nginx 0a693112c443ce4fb21bc57a26d67f0648b9415e052f929be7e06701f5f3ca2d $ sudo docker images REPOSITORY TAG IMAGE ID CREATED VIRTUAL SIZE ubuntu nginx 0a693112c443 9 seconds ago 153.1 MB ubuntu 12.04 0a51fcc173d8 2 days ago 111 MB ②用dockerfile来创建镜像 首先创建一个Dockerfile: $ cat Dockerfile yet another ngnix FROM ubuntu:12.04 MAINTAINER feisky feisky@root RUN apt-get update RUN apt-get -y install nginx put my local site to /var/www ADD index.html /var/www/html/ expose httpd port EXPOSE 80 the command to run CMD ["/usr/sbin/nginx"] $ sudo docker build -t 'ubuntu:www' . ③导入已有镜像 要从本地文件系统导入一个镜像,可以使用openvz(容器虚拟化的先锋技术)的模板来创建: openvz的模板下载地址为http://openvz.org/Download/templates/precreated $ sudo cat ubuntu-14.04-x86_64-minimal.tar.gz |docker import - ubuntu:14.04 ④导入docker save保存的镜像 $ sudo docker save -o ubuntu_12.04.tar ubuntu:12.04 $ sudo docker load --input ubuntu_14.04.tar $ sudo docker load < ubuntu_14.04.tar #同上 ⑤导入已保存的容器 保存一个容器的方法 $ sudo docker ps -a CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES 7691a814370e ubuntu:14.04 "/bin/bash" 36 hours ago Exited (0) 21 hours ago test $ sudo docker export 7691a814370e > ubuntu.tar 重新作为镜像导入进来 $ cat ubuntu.tar | sudo docker import - test/buntu:v1.0 注:用户既可以使用docker load来导入镜像存储文件到本地镜像库,也可以使用docker import来导入一个容器快照到本地镜像库。这两者的区别在于容器快照文件将丢弃所有的历史记录和元数据信息(即仅保存容器当时的快照状态),而镜像存储文件将保存完整记录,体积也要大。此外,从容器快照文件导入时可以重新指定标签等元数据信息。 镜像制作好后,可以通过docker push命令,把自己创建的镜像上传到仓库中来共享 $ sudo docker push ubuntu 如果一个镜像不需要了,可以删除它: $ sudo docker rmi e01000c7bac8 容器管理 容器是独立运行的一个或一组应用,以及它们的运行态环境。 启动容器 下面的命令输出一个"Hello World",之后终止容器: $ sudo docker run ubuntu:nginx /bin/echo 'hello world' hello world 下面的命令则启动一个bash终端,可以让用户进行交互。 $ sudo docker run -t -i ubuntu:12.04 /bin/bash root@af8bae53bdd3:/# 其中,-t选项让Docker分配一个伪终端(pseudo-tty)并绑定到容器的标准输入上, -i则让容器的标准输入保持打开。 对于已经停止的容器,可以用start命令重新启动: $ sudo docker start 7fb349365baf 7fb349365baf $ sudo docker ps CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES 7fb349365baf ubuntu:nginx /bin/bash 29 minutes ago Up 6 seconds goofy_ritchie0 docker run的一些有用参数: ①-d:以守护进程形式运行容器 ② 其他命令: nsenter --target $PID --mount --uts --ipc --net --pid连接容器终端 docker attach连接容器终端 docker stop来终止一个运行中的容器 docker restart命令会将一个运行态的容器终止,然后再重新启动它 docker logs获取容器的输出信息 docker ps查看正在运行的容器 docker ps查看正在运行和已经停止的容器 注意:当多个窗口同时attach到同一个容器的时候,所有窗口都会同步显示。当某个窗口因命令阻塞时,其他窗口也无法执行操作了。 仓库管理 对于默认Docker Hub仓库,通过执行docker login命令来输入用户名、密码和邮箱来完成注册和登录。 注册成功后,本地用户目录的.dockercfg中将保存用户的认证信息。 可以通过sudo docker search centos来搜索镜像,通过sudo docker pull centos来下载镜像。 https://registry.hub.docker.com/builds/add/提供的自动构建功能对于需要经常升级程序的镜像比较有用,目前仅支持Github和BitBucket。 私有仓库的搭建 ① $ sudo docker run -d -p 5000:5000 registry ② $ sudo pip install docker-registry ③ $ cp config/config_sample.yml config/config.yml ④ $ sudo gunicorn --access-logfile - --error-logfile - -k gevent -b 0.0.0.0:5000 -w 4 --max-requests 100 docker_registry.wsgi:application 如何向私有仓库上传镜像 ① $ sudo docker tag ba58 192.168.7.26:5000/test ② $ sudo docker push 192.168.7.26:5000/test 通过$ curl http://192.168.7.26:5000/v1/search可以查询私有仓库的镜像,通过sudo docker pull 192.168.7.26:5000/test可以下载私有仓库的镜像。 数据管理 创建一个web容器,并加载一个数据卷到容器的/webapp目录: $ sudo docker run -d -P --name web -v /webapp training/webapp python app.py 挂载一个主机目录作为数据卷: $ sudo docker run -d -P --name web -v /src/webapp:/opt/webapp training/webapp python app.py Docker挂载数据卷的默认权限是读写,用户也可以通过:ro指定为只读: $ sudo docker run -d -P --name web -v /src/webapp:/opt/webapp:ro training/webapp python app.py 也可以挂载一个本地主机文件作为数据卷:$ sudo docker run --rm -it -v ~/.bash_history:/.bash_history ubuntu /bin/bash,注意这会导致报错误信息,最好还是直接挂载文件的父目录 如果你有一些持续更新的数据需要在容器之间共享,最好创建数据卷容器。数据卷容器,其实就是一个正常的容器,专门用来提供数据卷供其它容器挂载的: $ sudo docker run -d -v /dbdata --name dbdata training/postgres echo Data-only container for postgres 然后,在其他容器中使用--volumes-from来挂载dbdata容器中的数据卷。 $ sudo docker run -d --volumes-from dbdata --name db1 training/postgres $ sudo docker run -d --volumes-from dbdata --name db2 training/postgres 网络管理 当Docker启动时,会自动在主机上创建一个docker0虚拟网桥,实际上是Linux的一个bridge,可以理解为一个软件交换机。它会在挂载到它的网口之间进行转发。 同时,Docker随机分配一个本地未占用的私有网段(在RFC1918中定义)中的一个地址给docker0接口。比如典型的172.17.42.1,掩码为255.255.0.0。此后启动的容器内的网口也会自动分配一个同一网段. ① 端口映射: 使用docker port 来查看当前映射的端口配置 $ sudo docker run -P ... #随机映射一个49000~49900的端口到内部容器开放的网络端口 $ sudo docker run -d -p 5000:5000 #映射到指定端口 $ sudo docker run -d -p 127.0.0.1:5000:5000 #映射到指定HOST+端口 $ sudo docker run -d -p 127.0.0.1::5000 #映射到指定HOST,端口随机生成 $ sudo docker run -d -p 127.0.0.1:5000:5000/udp #映射到指定UDP端口 ② 容器互联 启动容器时指定容器名称: $ sudo docker run -d --name db training/postgres 根据名称连接db容器 $ sudo docker run -d -P --name web --link db:db training/webapp python app.py 注意:--link参数的格式为--link name:alias,其中name是要链接的容器的名称,alias是这个连接的别名;如果名称未知,可以通过下面的命令查询: $ sudo docker inspect -f "{{ .Name }}" aed84ee21bde Docker在两个互联的容器之间创建了一个安全隧道,而且不用映射它们的端口到宿主主机上。在启动db容器的时候并没有使用-p和-P标记,从而避免了暴露数据库端口到外部网络上。 Docker 通过2种方式为容器公开连接信息: 其一是环境变量: $ sudo docker run --rm --name web2 --link db:db training/webapp env . . . DB_NAME=/web2/db DB_PORT=tcp://172.17.0.5:5432 DB_PORT_5000_TCP=tcp://172.17.0.5:5432 DB_PORT_5000_TCP_PROTO=tcp DB_PORT_5000_TCP_PORT=5432 DB_PORT_5000_TCP_ADDR=172.17.0.5 . . . 其二是hosts: $ sudo docker run -t -i --rm --link db:db training/webapp /bin/bash root@aed84ee21bde:/opt/webapp # cat /etc/hosts 172.17.0.7 aed84ee21bde . . . 172.17.0.5 db ③ 其他选项 只有在Docker服务启动的时候才能配置: -b BRIDGE or --bridge=BRIDGE --指定容器挂载的网桥 --bip=CIDR --定制docker0的掩码 -H SOCKET... or --host=SOCKET... --Docker服务端接收命令的通道 --icc=true|false --是否支持容器之间进行通信 --ip-forward=true|false --请看下文容器之间的通信 -iptables=true|false --禁止Docker添加iptables规则 --mtu=BYTES --容器网络中的MTU 既可以在启动服务时指定,也可以Docker容器启动(docker run)时候指定: --dns=IP_ADDRESS... --使用指定的DNS服务器 --dns-search=DOMAIN... --指定DNS搜索域 只有在docker run执行时使用: -h HOSTNAME or --hostname=HOSTNAME --配置容器主机名 --link=CONTAINER_NAME:ALIAS --添加到另一个容器的连接 --net=bridge|none|container:NAME_or_ID|host --配置容器的桥接模式 -p SPEC or --publish=SPEC --映射容器端口到宿主主机 -P or --publish-all=true|false --映射容器所有端口到宿主主机 本文转自feisky博客园博客,原文链接:http://www.cnblogs.com/feisky/p/4011843.html,如需转载请自行联系原作者

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

ADB笔记

Android Debug Bridge 本文以MAC OS为例 ADB版本 $ ./adb version 如果出现 $ ./adb version -bash: ./adb: No such file or directory 说明你还没有配置环境变量 你可以来到ADB的目录下再执行命令,目录在 $ cd Library/Android/sdk/platform-tools/ 也可以自行google配置一下环境变量 这样就可以在任意目录使用ADB指令 终端开启Debug模式 在设置中找到开发者选项,并开启debug模式 开发者选项默认是隐藏状态,如何开启自行google 查看设备 $ ./adb devices List of devices attached 4f0a34ac device 5F6F7108 device 前面是序列号,后面是设备状态 状态 说明 device 实例现在已连接到 adb 服务器。请注意,此状态并不表示 Android 系统已完全启动且可以运行,因为在此实例连接到 adb 时系统仍在启动。不过,在启动后,这将是模拟器/设备实例的正常运行状态。 offline 实例未连接到 adb 或不响应。 no device 未连接模拟器/设备。 发送指令到指定终端 $ ./adb -s 序列号 指令 参数 说明 注释 -s 序列号 将 adb 命令发送至以其 adb 分配的序列号命名的特定模拟器/设备实例(如“emulator-5556”)。 -d 将 adb 命令发送至唯一连接的 USB 设备。 如果连接了多个 USB 设备,将返回错误。 -e 将 adb 命令发送至唯一运行的模拟器实例。 如果有多个模拟器实例在运行,将返回错误。 安装应用 多个设备 $ ./adb -s 序列号 install xxx/xxx/xxx.apk 单个真机设备 $ ./adb -d install xxx/xxx/xxx.apk 单个模拟器 $ ./adb -e install xxx/xxx/xxx.apk 从PC端复制文件到终端 $ ./adb pull `终端文件绝对路径` `PC端文件绝对路径` e.g.:将手机SD下的demo.mp4文件拷贝到桌面 $ ./adb -d pull /sdcard/demo.mp4 /Users/kongqingwei/Desktop [100%] /sdcard/demo.mp4 从终端复制文件到PC端 $ ./adb -d push `PC端文件绝对路径` `终端文件绝对路径` e.g.:将桌面的图片拷贝到终端SD卡 $ ./adb -d push /Users/kongqingwei/Desktop/ic_launcher.png /sdcard/ic_launcher.png [100%] /sdcard/ic_launcher.png 关闭ADB 终止 adb 服务器进程。 当adb无响应的时候可以关闭adb再重启,得以解决问题。 $ ./adb kill-server 启动ADB 检查 adb 服务器进程是否在运行,如果未运行则启动它。 $ ./adb start-server * daemon not running. starting it now on port 5037 * * daemon started successfully * 进入终端shell $ ./adb [-d|-e|-s `序列号`] shell `shell指令` 或 $ ./adb [-d|-e|-s `序列号`] shell e.g. $ ./adb -d shell ls acct bin cache config d data default.prop …… 或 $ ./adb -d shell shell@mocha:/ $ 退出终端shell shell@mocha:/ $ exit 打印Logcat $ ./adb logcat [<option>] ... [<filter-spec>] ... 或 $ ./adb shell [<option>] ... [<filter-spec>] ... $ logcat 详细参数 截屏 截屏保存到/sdcard/screen.png $ ./adb -d shell screencap /sdcard/screen.png 录屏 按 Control + C 停止屏幕录制,否则,到三分钟或 –time-limit 设置的时间限制时,录制将自动停止。 $ ./adb -d shell screenrecord /sdcard/demo.mp4 ^C 参数 说明 –help 显示命令语法和选项 –size widthxheight 设置视频大小:1280x720。默认值是设备的原生显示分辨率(如果支持),如果不支持,则使用 1280x720。为实现最佳结果,请使用设备的 Advanced Video Coding (AVC) 编码器支持的大小。 –bit-rate rate 设置视频的视频比特率(以兆比特每秒为单位)。默认值为 4Mbps。您可以增加比特率以提升视频质量,但这么做会导致影片文件变得更大。以下示例将录制比特率设为6Mbps:screenrecord --bit-rate 6000000 /sdcard/demo.mp4 –time-limit time 设置最大录制时长(以秒为单位)。默认值和最大值均为 180(3 分钟)。 –rotate 将输出旋转 90 度。此功能是实验性的。 –verbose 显示命令行屏幕上的日志信息。如果您不设置此选项,则运行时此实用程序不会显示任何信息。 Activity Manager (am) 软件包管理器 (pm)

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

【开发者笔记】MQTT python测试笔记

MQTT是基于订阅/发布的物联网协议。 python测试需要一个发送进程和接收进程,即一个发送客户端和一个接收客户端,如果这两个客户端工作在同一个topic下,那么就能进行消息互通了。 服务器用“iot.eclipse.org”就好了,避免了自己搭建服务器,然后流程还可以跑通。 发送客户端代码: import paho.mqtt.client as mqtt import paho.mqtt.publish as publish idx = 0#往paho/temperature 一直发送内容 while True: print("send success") publish.single("paho/temperature", payload="this is message:%s"%idx, hostname="iot.eclipse.org", client_id="lora1", # qos = 0, # tls=tls, port=1883, protocol=mqtt.MQTTv311) idx += 1 接收客户端代码: import paho.mqtt.client as mqtt # The callback for when the client receives a CONNACK response from the server. def on_connect(client, userdata, flags, rc): print("Connected with result code "+str(rc)) # The callback for when a PUBLISH message is received from the server. def on_message(client, userdata, msg): #在这里处理业务逻辑 print(msg.topic+" "+str(msg.payload)) client = mqtt.Client() client.on_connect = on_connect client.on_message = on_message client.connect("iot.eclipse.org", 1883, 60)#订阅频道 client.subscribe("paho/temperature") # Blocking call that processes network traffic, dispatches callbacks and # handles reconnecting. # Other loop*() functions are available that give a threaded interface and a # manual interface. client.loop_forever() 然后运行两个客户端,就可以在接收端收到消息了。 MQTT服务器不负责存储数据,需要编写额外的接收客户端来接收数据、分析、入库等。 MQTT服务器用的是iot.eclipse.org,如果碰巧两个人在用同一个频道,那可能收到别人的消息哦~ 如果要搭建自己的MQTT服务器,那么回头再说。 玩一玩就好了,不要给服务器增加太多负担哟~ 参考资料: paho-qtt说明文档 黑夜给了我黑色的眼睛,我却用它寻找光明

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

Flume学习笔记

Flume [fluːm] (引水槽、运河) 王洪亮 2021/05/07 15:05 1. 简介 大数据技术解决的三个问题,海量数据的传输、存储和计算。Flume属于传输框架。专门传输日志的,多媒体不行。 2.安装 官网 http://flume.apache.org/ 下载apache-flume-1.9.0-bin.tar.gz,下载后解压即可,类似tomcat中间件,需要改配置文件 , 把apache-flume-1.9.0-bin/conf/flume-env.sh.template改成flume-env.sh,里面配置jdk地址,安装完毕。 export JAVA_HOME=D:\sde\Java\jdk1.8.0_131 3.Hello Flume样例 flume是按任务启动的,这个任务在flume这里叫agent,每个agent分为三块,source(对接数据来源)、channel(中间缓冲区)、sink(数据去向)。任务过程简述就是从一个地方采集数据后送到其他地方。 此次案例source数据采集自端口,通过channel再经过sink将数据输出到终端屏幕上。 开始编写Hello Flume样例(win环境) 编写配置文件,参考官网例子 习惯上把任务的配置文件和conf区分开,这里新建jobs目录下新建netcat-flume-logger.conf,命名格式最好区分出source和sink的类型 #完整的conf分几块,变量声明区、source区、interceptor区、channelSelector区、channel区、sink区、关系绑定区 #变量声明区(必须) a1指的是agent的变量名,启动时指定 #分别声明 sources、channels、sinks、sinkgroups等 a1.sources = r1 a1.channels = c1 a1.sinks = k1 #source区(必须) 指定source类型 具体类型需要看官网文档source 部分 a1.sources.r1.type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444 #interceptor区(可选),代码自写,配合multiplexing channelSelector使用 #channelSelector区(可选),不写默认是replicating,把数据发给每个sink a1.sources.r1.selector.type=replicating #channel区(必须) 指定channel类型 具体类型需要看官网文档channel部分 a1.channels.c1.type = memory #sink处理器区,声明sink组,选择处理器是故障转移还是负载均衡,不写就是默认 #sink区(必须) 指定sink类型 具体类型需要看官网文档sink部分 a1.sinks.k1.type = logger #关系绑定区(必须) a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 启动flume的agent #win环境启动命令 进入到bin目录 cmd flume-ng agent --conf ../conf --conf-file ../jobs/netcat-flume-logger.conf --name a1 -property flume.root.logger=INFO,console #简写 flume-ng agent -c ../conf -f ../jobs/netcat-flume-logger.conf -n a1 -property flume.root.logger=INFO,console #linux环境启动命令 那个logger不一样 #-Dflume.root.logger=INFO,console 启动结果显示端口已监听 [INFO - org.apache.flume.source.NetcatSource.start(NetcatSource.java:166)] Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/127.0.0.1:44444] 通过netcat工具向端口发送数据 没有netcat telnet 也可以 nc localhost 44444 > Hello Flume OK 查看flume的日志输出 [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 48 65 6C 6C 6F 20 46 6C 75 6D 65 Hello Flume } 4.Source种类 拉出几个常用的。其他的查看官方文档 1.Avro Source Flume集群使用,多个Fulme连接需要使用Avro的source和sink连接,自身相当于服务端,需要优先启动 #相关source区 写法 a1.sources.r1.type=avro a1.sources.r1.bind=0.0.0.0 a1.sources.r1.port=4141 2. Exec Source 用命令监控某个文件的变化,将变化的内容传送走,适用于监控一个实时变化的日志文件,缺点是不能断点续传,容易数据丢失或是数据重复 #相关source区 写法 a1.sources.r1.type=exec #后跟 监控文件的命令 a1.sources.r1.command=tail -F /var/log/secure 3. Spooling Directory Source 监控本地文件夹,文件夹中新增文件时,会将整个文件内容取走,并加后缀做标记(.COMPLETED),表示已经取过了,标记后的文件不在处理,内容变化也不管。适用于批量上传,缺点是不能适应动态变化的文件 #相关source区 写法 a1.sources.r1.type=spooldir #后跟监控文件的目录 a1.sources.r1.spoolDir=/var/log/apache/flumeSpool 4. Taildir Source 可以监控不同文件夹下的不同文件。支持断点续传 #相关source区 写法 a1.sources.r1.type = TAILDIR #指定保存断点续传的文件 a1.sources.r1.positionFile = /var/log/flume/position.json #指定文件组 可以多个 a1.sources.r1.filegroups = f1 f2 #指定 单个文件 a1.sources.r1.filegroups.f1 = /var/log/test1/example.log #指定多个文件 * 不能出现在第一位 a1.sources.r1.filegroups.f2 = /var/log/test2/.*log.* 5. NetCat TCP Source 监听TCP端口用的,还有个UDP Source 就type不一样 #相关source区 写法 udp 的type是netcatudp a1.sources.r1.type=netcat a1.sources.r1.bind=0.0.0.0 a1.sources.r1.port=6666 5.Channel种类 常用的Memory Channel和File Channel 1. Memory Channel 将数据缓存在内存中,快,闪断丢数据 #channel区 a1.channels.c1.type=memory 2. File Channel 将数据缓存在本地文件中,慢点,稳定 #channel区 a1.channels.c1.type=file #数据存放的目录,可以有多个目录,逗号分割 a1.channels.c1.dataDirs=/mnt/flume/data 6.Sink种类 常用的拉出 1. Avro Sink Flume集群收尾连接,相对source 而言这个算客户端,去绑定服务端IP和端口 #sink 区 a1.sinks.k1.type = avro a1.sinks.k1.hostname = 10.10.10.10 a1.sinks.k1.port = 4545 2. Logger Sink 日志打印在终端屏幕上,需要agent启动的时候搭配对应命令 问题:中文没打出来;maxBytesToLog这玩意不好使 #sink区 #-property flume.root.logger=INFO,console #-Dflume.root.logger=INFO,console a1.sinks.k1.type=logger 3. File Roll Sink 日志输出到本地文件,问题是就算没有接收到数据,也会不停的生成新的空文件 #sink区 a1.sinks.k2.type = file_roll #指定生成文件夹路径,文件名会自动生成 a1.sinks.k2.sink.directory = ../data/ #指定新文件生成频率,30s a1.sinks.k2.sink.rollInterval = 30 7.Channel Selector 这个就俩,算自定义的三个 1. Replicating Channel Selector (default) 默认的,可以不写,把数据发给所有sink #channel selector 区 a1.sources.r1.selector.type=replicating 2. Multiplexing Channel Selector 根据拦截器规则,选择发送到哪个sink #channel selector 区 a1.sources.r1.selector.type = multiplexing #这个关键变量,拦截器里写的,写什么是什么,不一定是state a1.sources.r1.selector.header = state #根据这个state的值选择sink,如果header中的state=CZ就发个c1 a1.sources.r1.selector.mapping.CZ = c1 # 等于US就给c2 c3 a1.sources.r1.selector.mapping.US = c2 c3 8.Sink Processor 目前算默认的三种,默认的数据给谁,谁就发走 1. Failover Sink Processor 故障转移,根据优先级,可着最高优先级的用,挂了再用优先级低的 #sink处理器区 #这个算声明,可以拿到声明区 a1.sinkgroups = g1 a1.sinkgroups.g1.sinks = k1 k2 #选择处理器类型为 故障转移 a1.sinkgroups.g1.processor.type = failover #指定sink组成员优先级 a1.sinkgroups.g1.processor.priority.k1 = 5 #最高等级k2一直使用,直到挂了为止,才能轮到k1 a1.sinkgroups.g1.processor.priority.k2 = 10 2. Load balancing Sink Processor 负载均衡,可选随机或是均衡发送 #sink处理器区 #这个算声明,可以拿到声明区 a1.sinkgroups = g1 a1.sinkgroups.g1.sinks = k1 k2 #选择处理器类型为 负载均衡 a1.sinkgroups.g1.processor.type = load_balance #指定模式为随机发送,还可以选round_robin均衡发送 a1.sinkgroups.g1.processor.selector = random 9.自定义Inteceptor idea新建maven项目开整 \> hello 1. 引入依赖 <dependency> <groupId>org.apache.flume</groupId> <artifactId>flume-ng-core</artifactId> <version>1.9.0</version> </dependency> 2. 写自己的拦截器 package com.flume; import org.apache.flume.Context; import org.apache.flume.Event; import org.apache.flume.interceptor.Interceptor; import java.util.HashMap; import java.util.List; import java.util.Map; public class MyInterceptor implements Interceptor { @Override public void initialize() { System.out.println("---------initialize---------"); } @Override public Event intercept(Event event) { String body = new String(event.getBody()); Map<String, String> map = new HashMap<>(); if(body.contains("hello")){ map.put("myStatus","up"); event.setHeaders(map); }else{ map.put("myStatus","down"); event.setHeaders(map); } return event; } @Override public List<Event> intercept(List<Event> list) { return null; } @Override public void close() {} public static class MyBuilder implements Interceptor.Builder{ @Override public Interceptor build() { return new MyInterceptor(); } @Override public void configure(Context context) {} } } 3. 项目打个jar包放到Flume的lib目录下 4. 编写Flume配置文件 netcat-flume-doubleLogger.conf,本来想两个sink都发到logger的,后来发现终端上分不出哪个是哪个,把其中一个sink改到落到本地文件了 #变量区 a1.sources = r1 a1.channels = c1 c2 a1.sinks = k1 k2 #source区 a1.sources.r1.type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444 #intercetor区 a1.sources.r1.interceptors = i1 a1.sources.r1.interceptors.i1.type = com.flume.MyInterceptor$MyBuilder #channel selector 区 a1.sources.r1.selector.type = multiplexing #我代码里定义的是 myStatus a1.sources.r1.selector.header = myStatus #myStatus=up就发个c1 a1.sources.r1.selector.mapping.up = c1 # 等于down就给c2 a1.sources.r1.selector.mapping.down = c2 #channel区 a1.channels.c1.type = memory a1.channels.c2.type = memory #sink区 a1.sinks.k1.type = logger a1.sinks.k2.type = file_roll a1.sinks.k2.sink.directory = ../data/ #绑定区 a1.sources.r1.channels = c1 c2 a1.sinks.k1.channel = c1 a1.sinks.k2.channel = c2 5. 启动 flume-ng agent --conf ../conf --conf-file ../jobs/netcat-flume-doubleLogger.conf --name a1 -property flume.root.logger=INFO,console 6. 测试效果 启动netcat发送命令 > nc localhost 44444 6351 OK hello OK 一个在终端屏幕上有显示,一个在本地文件中有显示。 10自定义Source 1. 依赖 <dependency> <groupId>org.apache.flume</groupId> <artifactId>flume-ng-core</artifactId> <version>1.9.0</version> </dependency> 2. 代码 package com.flume; import org.apache.flume.Context; import org.apache.flume.Event; import org.apache.flume.PollableSource; import org.apache.flume.conf.Configurable; import org.apache.flume.event.SimpleEvent; import org.apache.flume.source.AbstractSource; public class MySource extends AbstractSource implements Configurable, PollableSource { private String aaa; private String bbb; /** * 定义配置文件中的变量 * @param context */ @Override public void configure(Context context) { aaa = context.getString("aaa"); //没取到值 就给默认值 bbb = context.getString("bbb", "GangZi"); } @Override public Status process(){ Status status = null; try { //创建个事件 Event e = new SimpleEvent(); String content = aaa+"->"+bbb+": "+Math.random(); e.setBody(content.getBytes("utf-8")); //把event传给channel getChannelProcessor().processEvent(e); status = Status.READY; } catch (Throwable t) { status = Status.BACKOFF; if (t instanceof Error) { throw (Error)t; } } try { //睡两秒 要不刷的太快了 Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } return status; } @Override public long getBackOffSleepIncrement() { return 0; } @Override public long getMaxBackOffSleepInterval() { return 0; } } 3. 写配置文件custom-flume-logger.conf #变量区 a1.sources = r1 a1.channels = c1 a1.sinks = k1 #source区 a1.sources.r1.type = com.flume.MySource a1.sources.r1.aaa = LiLei #a1.sources.r1.bbb = HanMeiMei #channel区 a1.channels.c1.type = memory #sink区 a1.sinks.k1.type = logger a1.sinks.k1.maxBytesToLog = 18 #绑定区 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 4. 启动命令 flume-ng agent --conf ../conf --conf-file ../jobs/custom-flume-logger.conf --name a1 -property flume.root.logger=INFO,console 5. 查看效果 [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 4C 69 4C 65 69 2D 3E 47 61 6E 67 5A 69 3A 20 30 LiLei->GangZi: 0 } 11自定义Sink 1. 依赖 <dependency> <groupId>org.apache.flume</groupId> <artifactId>flume-ng-core</artifactId> <version>1.9.0</version> </dependency> 2. 代码 package com.flume; import org.apache.flume.*; import org.apache.flume.conf.Configurable; import org.apache.flume.sink.AbstractSink; public class MySink extends AbstractSink implements Configurable { private String ccc; private String ddd; @Override public void configure(Context context) { ccc = context.getString("ccc"); ddd = context.getString("ddd","xxxxxxxx"); } @Override public Status process() throws EventDeliveryException { Status status = null; Channel ch = getChannel(); Transaction txn = ch.getTransaction(); txn.begin(); try { Event event = ch.take(); if(event == null){ txn.rollback(); status = Status.BACKOFF; }else { byte[] body = event.getBody(); String content = new String(body); System.out.println("ccc="+ccc); System.out.println("content="+content); System.out.println("ddd="+ddd); txn.commit(); status = Status.READY; } } catch (Exception e) { e.printStackTrace(); txn.rollback(); status = Status.BACKOFF; }finally { txn.close(); } return status; } } 3. flume配置文件netcat-flume-custom.conf #变量区 a1.sources = r1 a1.channels = c1 a1.sinks = k1 #source区 a1.sources.r1.type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444 #channel区 a1.channels.c1.type = memory #sink区 a1.sinks.k1.type = com.flume.MySink a1.sinks.k1.ccc =--hello-- a1.sinks.k1.ddd =--moto-- #绑定区 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 4. 启动agent flume-ng agent --conf ../conf --conf-file ../jobs/netcat-flume-custom.conf --name a1 -property flume.root.logger=INFO,console 5. 客户端发送命令 telnet localhost 44444 > 1 OK 6. flume终端打印 ccc=--hello-- content=1 ddd=--moto-- 12.Flume监控工具 下面的实验,windows环境下怎么跟多个参数 不知道 1.Http监控,json形式文本 使用这种监控方式,只需要在启动flume的时候在启动参数上面加上监控配置,例如这样: bin/flume-ng agent --conf conf --conf-file conf/flume_conf.properties --name collect -Dflume.monitoring.type=http -Dflume.monitoring.port=1234 其中-Dflume.monitoring.type=http表示使用http方式来监控,后面的-Dflume.monitoring.port=1234表示我们需要启动的监控服务的端口号为1234,这个端口号可以自己随意配置。然后启动flume之后,通过http://ip:1234/metrics就可以得到flume的一个json格式的监控数据。 2.ganglia监控,图形化界面 这种监控方式需要先安装ganglia然后启动ganglia,然后再启动flume的时候加上监控配置,例如: bin/flume-ng agent --conf conf --conf-file conf/producer.properties --name collect -Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=ip:port 其中-Dflume.monitoring.type=ganglia表示使用ganglia的方式来监控,而-Dflume.monitoring.hosts=ip:port表示ganglia安装的ip和启动的端口号。 flume监控还可以使用zabbix,但是这种方式需要在flume源码中添加监控模块,相对比较麻烦,由于不是flume自带的监控方式,这里不讨论这种方式。 因此,flume自带的监控方式其实就是http、ganglia两种,http监控只能通过一个http地址访问得到一个json格式的监控数据,而ganglia监控是拿到这个数据后用界面的方式展示出来了,相对比较直观。

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

HTML 学习笔记

基本框架 <!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <meta http-equiv="X-UA-Compatible" content="ie=edge"> <title>标题</title> </head> <body> </body> </html> 可打开 .html 文件,直接输入 html:5 调出 语句 一些规范: 标签使用小写,元素必须闭合 空元素要加斜杠以闭合 eg. <br /> 不使用语义化,所有样式都存放于 CSS 中,内容与样式分离 <!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <meta http-equiv="X-UA-Compatible" content="ie=edge"> <title>标题</title> </head> <body> <h1>一级标题</h1> <h2>二级标题</h2> <p>段落</p> <!--换行符--> <br /> <!--分割线--> <hr /> <!--列表,可嵌套--> <!--有序列表--> <ol> <li>第一项</li> <li>第二项</li> </ol> <!--无序列表--> <ul> <li>第一项</li> <li>第二项</li> </ul> <!--链接--> <a href="https://www.google.com/">链接显示的文本</a> <!--链接到页面特定位置,使用 ID 特性--> <a href="#top">回到顶部</a> <p id="top">顶部</p> <!--链接到其他页面的特定位置--> <a href="http://wiki-power.com/#top">跳转到站外页面的某个位置</a> <!--图像--> <img src="/xx.png" alt="无法加载时的文字说明" /> <!--表格--> <table> <!--第一行--> <tr> <!--第一列--> <th></th> <!--第二列--> <th scope="col">周六</th> <!--第三列--> <th scope="col">周日</th> </tr> <!--第二行--> <tr> <th scope="row">数量</th> <td>120</td> <td>135</td> </tr> <!--第三行--> <tr> <th scope="row">收益</th> <!--跨列 colspan,跨行 rowspan--> <td colspan="2">500</td> </tr> </table> <!--表单,待补充--> <!--iframe,待补充--> <!--flash/视频/音频,待补充--> </body> </html> 参考与致谢 HTML 教程 | 菜鸟教程 HTML 30 分钟入门教程 HTML - head 头部浅析 文章作者:Power Lin 原文地址:https://wiki-power.com 版权声明:文章采用 CC BY-NC-SA 4.0 协议,转载请注明出处。

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

Flink学习笔记

Flink 基础 Flink特性 流式计算是大数据计算的痛点,第1代实时计算引擎Storm对Exactly Once 语义和窗口支持较弱,使用的场景有限且无法支持高吞吐计算;Spark Streaming 采用“微批处理”模拟流计算,在窗口设置很小的场景中有性能瓶颈,Spark 本身也在尝试连续执行模式(Continuous Processing),但进展缓慢。 Flink是一个低延迟、高吞吐的实时计算引擎,其利用分布式一致性快照实现检查点容错机制,并实现了更好的状态管理,Flink可在毫秒级的延迟下处理上亿次/秒的消息或者事件,同时提供了一个Exactly-once的一致性语义,保证了数据的正确性,使得Flink可以提供金融级的数据处理能力,总结其高级特性包括CSTW(CheckPoint,Statue,Time,windows) Flink和Spark对比 设计思路 Spark的技术理念是基于批来模拟流,微批处理的延时较高(无法优化到秒以下的数量级),且无法支持基于event_time的时间窗口做聚合逻辑。Flink和spark相反,它基于流计算来模拟批计算,更切合数据的生成方式,技术上有更好的扩展性。 状态管理 流处理任务要对数据进行统计,如Sum, Count, Min, Max,这些值是需要存储的,因为要不断更新,这些值或者变量就可以理解为一种状态,如果数据源是在读取Kafka, RocketMQ,可能要记录读取到什么位置,并记录Offset,这些Offset变量都是要计算的状态。 Flink提供了内置的状态管理,可以把这些状态存储在Flink内部,而不需要把它存储在外部系统,这样做的好处: ① 降低了计算引擎对外部系统的依赖以及部署,使运维更加简单; ② 对性能带来了极大的提升:如果通过外部去访问如Redis , HBase 需要网络及RPC资源,如果通过Flink内部去访问,只通过自身的进程去访问这些变量。 同时Flink会定期将这些状态做Checkpoint持久化,把Checkpoint存储到一个分布式的持久化系统中,比如HDFS,这样当Flink的任务出现任何故障时,它都会从最近的一次Checkpoint将整个流的状态进行恢复,然后继续运行它的流处理,对用户没有任何数据上的影响。 Flink 初探 设计架构 Flink是一个分层的架构系统,每一层所包含的组件都提供了特定的抽象,用来服务于上层组件,Flink的分层体现有四层,分别是Deploy层、core层、API层/Libraries层,其中Deploy层主要涉及的是Flink的部署模式及同资源调度组件的交互模式,Core层提供了支持Flink计算的全部核心实现,API层/Libraries层提供了Flink的API接口和基于API接口的特定应用的计算框架; Deploy层:该层主要涉及了Flink的部署模式,Flink支持多种部署模式:本地、集群(Standalone/YARN)、云(GCE/EC2),Standalone 部署模式与Spark类似; Runtime层:Runtime层提供了支持Flink计算的全部核心实现,比如:支持分布式Stream处理、Job Graph到Execution Graph的映射、调度 等,为上层API层提供基础服务。 API层:API层主要实现了面向无界Stream的流处理和面向Batch的批处理API,其中面向流处理对应DataStream API,面向批处理对应DataSet API。 Libraries层:该层也可以称为Flink应用框架层,根据API层的划分,在API层之上构建的满足特定应用的实时计算框架,也分别对应于面向流处理 和面向批处理两类。面向流处理支持:CEP(复杂事件处理)、SQL-like的操作(基于Table的关系操作);面向批处理支持:FlinkML(机器学习库)、Gelly(图处理)。 Flink on yarn Flink支持增量迭代,具有对迭代自行优化的功能,因此在on yarn上提交的任务性能略好于 Spark,Flink提供2种方式在yarn上提交任务:启动1个一直运行的 Yarn session(分离模式)和在 Yarn 上运行1个 Flink 任务(客户端模式); 分离模式:通过命令yarn-session.sh的启动方式本质上是在yarn集群上启动一个flink集群,由yarn预先给flink集群分配若干个container,在yarn的界面上只能看到一个Flink session with X TaskManagers的任务,并且只有一个Flink界面,可以从Yarn的Application Master链接进入; 客户端模式:通过命令bin/flink run -m yarn-cluster启动,每次发布1个任务,本质上给每个Flink任务启动了1个集群,yarn在任务发布时启动JobManager(对应Yarn的AM)和TaskManager,如果一个任务指定了n个TaksManager(-yn n),则会启动n+1个Container,其中一个是JobManager,发布m个应用,则有m个Flink界面,不同的任务不可能在一个Container(JVM)中,实现了资源隔离。 进入Flink的bin目录下运行./yarn-session.sh –help 查看帮助验证yarn是否成功配置,使用./yarn-session.sh –q 显示yarn所有nodeManager节点资源;部署On yarn模式的Flink只需要修改配置conf/flink-conf.yaml ,详细参数请参考官网:通用配置:Configuration,HA配置:High Availability (HA) 采用分离模式来启动Flink Yarn Session,提交后提示该yarn application成功提交到yarn并返回id,使用yarn application –kill application_id 来停止yarn上提交的任务; yarn-session.sh -n 3 -jm 700 -tm 700 -s 8 -nm FlinkOnYarnSession -d –st 可以直接提交自带的词频统计用例,验证on yarn模式是否配置成功: ~/bin/flinkrun-myarn-cluster-yn4-yjm2048-ytm2048~/flink/examples/batch/WordCount.jar 流程分析 分离模式:通过命令yarn-session.sh先启动集群,然后再提交作业,接着会向yarn申请一块空间后,资源永远保持不变。如果资源满了,下一个作业就无法提交,只能等到yarn中的其中一个作业执行完成后,释放了资源,下个作业才会正常提交。所有作业共享Dispatcher和ResourceManager;共享资源;适合规模小执行时间短的作业。 客户端模式: 通过命令bin/flink run -m yarn-cluster提交任务,每提交一个作业会根据自身的情况,都会单独向yarn申请资源,直到作业执行完成,一个作业的失败与否并不会影响下一个作业的正常提交和运行,适合规模大长时间运行的作业; DataStream DataStream是Flink的较低级API,用于进行数据的实时处理任务,可以将该编程模型分为DataSource、Transformation、Sink三个部分; DataSource 源是程序读取输入数据的位置,可以使用 StreamExecutionEnvironment.addSource(sourceFunction) 将源添加到程序,Flink 有许多预先实现的源函数,也可以通过实现 SourceFunction 方法自定义非并行源 ,或通过实现 ParallelSourceFunction 或扩展 RichParallelSourceFunction 自定义并行源。 有几个预定义的流数据源可从 StreamExecutionEnvironment 访问: 基于文件: readTextFile(path)#逐行读取文本文件(文件符合 TextInputFormat 格式),并作为字符串返回每一行。readFile(fileInputFormat,path)#按指定的文件输入格式(fileInputFormat)读取指定路径的文件。readFile(fileInputFormat,path,watchType,interval,pathFilter)#前两个方法的内部调用方法。根据给定文件格式(fileInputFormat)读取指定路径的文件。根据 watchType,定期监听路径下的新数据(FileProcessingMode.PROCESS_CONTINUOUSLY),或者处理当前在路径中的数据并退出(FileProcessingMode.PROCESS_ONCE),使用 pathFilter,可以进一步排除正在处理的文件。 基于Socket:socketTextStream 从 Socket 读取,元素可以用分隔符分隔。 基于集合: fromCollection(Seq)#用 Java.util.Collection 对象创建数据流,集合中的所有元素必须属于同一类型;fromCollection(Iterator)#用迭代器创建数据流。指定迭代器返回的元素的数据类型;fromElements(elements:_*)#从给定的对象序列创建数据流。所有对象必须属于同一类型;fromParallelCollection(SplittableIterator)#并行地从迭代器创建数据流。指定迭代器返回的元素的数据类型;generateSequence(from,to)#并行生成给定间隔的数字序列。 自定义:addSource 附加新的源函数。例如从 Apache Kafka 中读取,可以使用 addSource(new FlinkKafkaConsumer08<>(...))。请详细查看 连接器。 Transformation Transformation操作将1个或多个DataStream转换为新的DataStream,多个转换组合成复杂的数据流拓扑,如下图所示,DataStream会由不同的Transformation操作、转换、过滤、聚合成其他不同的流,从而完成业务要求; Map:DataStream -> DataStream,一个数据元生成一个新的数据元。将输入流的元素翻倍:dataStream.map { x => x * 2 } FlatMap:DataStream -> DataStream,一个数据元生成多个数据元(可以为0)。将句子分割为单词: dataStream.flatMap{str=>str.split("")} Filter:DataStream -> DataStream,每个数据元执行布尔函数,只保存函数返回 true 的数据元。过滤掉零值的过滤器: dataStream.filter{_!=0} KeyBy :DataStream -> KeyedStream,将流划分为不相交的分区。具有相同 Keys 的所有记录在同一分区。指定 key 的取值: dataStream.keyBy("someKey")//Keybyfield"someKey"dataStream.keyBy(0)//KeybythefirstelementofaTuple Reduce :KeyedStream -> DataStream,KeyedStream 元素滚动执行 Reduce。将当前数据元与最新的一个 Reduce 值组合作为新值发送。创建 key 的值求和:keyedStream.reduce { _ + _ } Aggregations :KeyedStream -> DataStream,应用于 KeyedStream 上的滚动聚合。 Window:KeyedStream -> WindowedStream,Windows 可以在已经分区的 KeyedStream 上定义。Windows 根据某些特征(例如,在最近5秒内到达的数据)对每个Keys中的数据进行分组。更多说明参考 Windows 或 译版。 dataStream.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5))) WindowAll :DataStream -> AllWindowedStream,Windows 也可以在 DataStream 上定义。在许多情况下,这是非并行转换。所有记录将收集在 windowAll 算子的一个任务中。 dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5))) Window Apply :WindowedStream -> DataStream 或 AllWindowedStream -> DataStream,将函数应用于整个窗口。一个对窗口数据求和: windowedStream.apply{WindowFunction}allWindowedStream.apply{AllWindowFunction} Window Reduce:WindowedStream -> DataStream,Reduce 函数应用于窗口并返回结果值。windowedStream.reduce { _ + _ } Aggregations on windows:WindowedStream -> DataStream,聚合窗口内容; Union :DataStream* -> DataStream,两个或多个数据流的合并,创建包含来自所有流的所有数据元的新流。如果将数据流与自身联合,则会在结果流中获取两次数据元。 dataStream.union(otherStream1,otherStream2,...) Window Join :DataStream,DataStream -> DataStream,Join 连接两个流,指定 Key 和窗口。 dataStream.join(otherStream).where(<keyselector>).equalTo(<keyselector>).window(TumblingEventTimeWindows.of(Time.seconds(3))).apply{...} Window CoGroup :DataStream,DataStream -> DataStream,CoGroup 连接两个流,指定 Key 和窗口。 dataStream.coGroup(otherStream).where(0).equalTo(1).window(TumblingEventTimeWindows.of(Time.seconds(3))).apply{} CoGroup 与 Join 的区别:CoGroup 会输出未匹配的数据,Join 只输出匹配的数据 Connect :DataStream,DataStream -> ConnectedStreams,连接两个有各自类型的数据流。允许两个流之间的状态共享。 someStream:DataStream[Int]=...otherStream:DataStream[String]=...valconnectedStreams=someStream.connect(otherStream) 可用于数据流关联配置流; CoMap, CoFlatMap :ConnectedStreams -> DataStream,作用域连接数据流(connected data stream)上的 map 和 flatMap: Split :DataStream -> SplitStream,将数据流拆分为两个或更多个流。 Select :SplitStream -> DataStream,从 SpliteStream 中选择一个流或多个流。 valeven=splitselect"even"valodd=splitselect"odd"valall=split.select("even","odd") Iterate :DataStream -> IterativeStream -> DataStream,将一个算子的输出重定向到某个先前的算子,在流中创建 feedback 循环。这对于定义不断更新模型的算法特别有用。以下代码以流开头并连续应用迭代体。大于0的数据元将被发送回 feedback,其余数据元将向下游转发。 Project:DataStream -> DataStream,作用于元组的转换,从元组中选择字段的子集。 DataStream<Tuple3<Integer,Double,String>>in=//[...]DataStream<Tuple2<String,Integer>>out=in.project(2,0); Sink Data Sink 消费 DataStream 并转发到文件,套接字,外部系统或打印到页面。Flink 带有各种内置输出格式,封装在 DataStreams 上的算子操作后面: writeAsText() / TextOutputFormat:按字符串顺序写入文件。通过调用每个元素的 toString() 方法获得字符串。 writeAsCsv(...) / CsvOutputFormat:将元组写为逗号分隔的形式写入文件。行和字段分隔符是可配置的。每个字段的值来自对象的 toString() 方法。 print() / printToErr():在标准输出/标准错误流上打印每个元素的 toString() 值。可以定义输出前缀,这有助于区分不同的打印调用。如果并行度大于1,输出也包含生成输出的任务的标识符。 writeUsingOutputFormat() / FileOutputFormat:自定义文件输出的方法和基类。支持自定义对象到字节的转换。 writeToSocket:将元素写入 Socket,使用 SerializationSchema 进行序列化。 addSink:调用自定义接收器函数。请详细查看 连接器。 DataStream 的 write*() 方法主要用于调试目的。他们没有参与 Flink checkpoint,这意味着这些函数通常具有至少一次的语义。刷新到目标系统的数据取决于 OutputFormat 的实现,并非所有发送到 OutputFormat 的数据都会立即显示在目标系统中。此外,在失败的情况下,这些记录可能会丢失。 要将流可靠、准确地传送到文件系统,请使用 flink-connector-filesystem。通过 .addSink(...) 方法的自定义实现,可以实现在 checkpoint 中精确一次的语义。 Time 流式数据处理最大的特点是数据具有时间属性特征,Flink根据时间产生的位置不同,将时间区分为三种概念:数据生成时间(Event_time)、事件接入时间(Ingestion_time)、事件处理时间(Processing_time),用户可以根据需要选择事件类型作为流式数据的时间属性,极大增强了数据处理的灵活性和准确性; Event_time:独立事件在产生它的设备上的发生时间,这个时间通常在到达Flink之前已经嵌入到生产数据中,因此时间顺序取决于事件产生的地方,和下游的数据处理系统的事件无关,需要在Flink中指定事件的时间属性或者设定时间提取器提取事件时间; Processing_time:指在操作算子计算过程中获取到的所在主机的时间,用户选择了Processing_time后,所有和时间相关的计算算子都直接使用其所在主机的系统时间,使用Processing_time的程序性能相对较高,延时相对较低,因为其所有操作不需要做任何时间上的对比和协调; Ingestion_time:指数据接入Flink系统的时间,依赖于Source Operator所在主机的系统时钟; 一般场景中选择event_time作为事件时间戳是最贴近生产的,但大多数情况下由于数据的延迟和乱序使用processing_time; Window窗口 Windows定义和分类 在流式计算中,数据持续不断的流入计算引擎,需要一个窗口限定计算范围,比如监控场景的近2分钟或者精准计算的每隔2分钟计算一次,窗口定义了该范围,辅助完成有界范围的数据处理; Flink的DataStream API将窗口抽象成独立的Operator,且支持很多窗口算子,每个窗口算子包含Window Assigner 、Windows Function、触发器、剔除器、时延设定等部分属性,其中Window Assigner 和 Windows Function是必须要指定的属性; Window Assigner用来决定某个元素被分配到哪个/哪些窗口中去;Trigger触发器决定了一个窗口何时能够被计算或清除,每个窗口都会拥有一个自己的Trigger; Evictor驱逐者在Trigger触发之后,在窗口被处理之前,Evictor(如果有Evictor的话)会用来剔除窗口中不需要的元素,相当于一个filter。 Flink支持多种窗口类型,按照驱动类型分为:时间驱动的Time Window(如每30秒钟)和数据驱动的Count Window(如每100个事件),按照窗口的滚动方式又可以分成:翻滚窗口(Tumbling Window,无重叠),滚动窗口(Sliding Window,有重叠)和会话窗口(Session Window,活动间隙),下图可以看出分类区别: Time Window 是根据时间对数据流进行分组的,且窗口机制和时间类型是完全解耦的,也就是说当需要改变时间类型时(三种时间)不需要更改窗口逻辑相关的代码,Time Window 中常见的即为Tumbling Time Window和Sliding Time Window; Count Window 是根据元素个数对数据流进行分组的,也包括Tumbling Count Window和Sliding Count Window; Windows实现 上图中的组件都位于一个算子(window operator)中,数据流源源不断地进入算子,每一个到达的元素都会被交给 WindowAssigner,WindowAssigner 会决定元素被放到哪个或哪些窗口(window),Window本身是一个ID标识符,其内部可能存储了一些元数据,如TimeWindow中有开始和结束时间,但是并不会存储窗口中的元素。窗口中的元素实际存储在 Key/Value State 中,key为Window,value为元素集合(或聚合值)。为了保证窗口的容错性,该实现依赖了 Flink 的 State 机制。 每一个窗口都拥有一个属于自己的 Trigger,Trigger上会有定时器,用来决定一个窗口何时能够被计算或清除,每当有元素加入到该窗口,或者之前注册的定时器超时了,那么Trigger都会被调用。Trigger的返回结果可以是 continue(不做任何操作),fire(处理窗口数据),purge(移除窗口和窗口中的数据),或者 fire + purge。一个Trigger的调用结果只是fire的话,那么会计算窗口并保留窗口原样,也就是说窗口中的数据仍然保留不变,等待下次Trigger fire的时候再次执行计算。一个窗口可以被重复计算多次知道它被 purge 了。在purge之前,窗口会一直占用着内存。 当Trigger fire了,窗口中的元素集合就会交给Evictor(如果指定了的话)。Evictor 主要用来遍历窗口中的元素列表,并决定最先进入窗口的多少个元素需要被移除。剩余的元素会交给用户指定的函数进行窗口的计算。如果没有 Evictor 的话,窗口中的所有元素会一起交给函数进行计算。 计算函数收到了窗口的元素(可能经过了 Evictor 的过滤),并计算出窗口的结果值,并发送给下游。窗口的结果值可以是一个也可以是多个。DataStream API 上可以接收不同类型的计算函数,包括预定义的sum(),min(),max(),还有 ReduceFunction,FoldFunction,还有WindowFunction。WindowFunction 是最通用的计算函数,其他的预定义的函数基本都是基于该函数实现的。 Flink 对于一些聚合类的窗口计算(如sum,min)做了优化,因为聚合类的计算不需要将窗口中的所有数据都保存下来,只需要保存一个result值就可以了。每个进入窗口的元素都会执行一次聚合函数并修改result值。这样可以大大降低内存的消耗并提升性能。但是如果用户定义了 Evictor,则不会启用对聚合窗口的优化,因为 Evictor 需要遍历窗口中的所有元素,必须要将窗口中所有元素都存下来。 Windows Function 在运用窗口计算时,Flink根据上有数据集是否是KeyedStream类型(数据是否按照Key分区),如果上游数据未分组则调用window()方法指定Windows Assigner,数据会根据Key在不同Task实例中并行计算,最后得出针对每个Key的统计结果,如果是Non-Keyed类型则调用WindowsAll()方法指定Windows Assigner,所有的数据都会在窗口算子中路由得到一个Task中计算,并得到全局统计结果; 定义完窗口分配器后,需要为每一个窗口指定计算逻辑,也就是Windows Function,Flink提供了四种类型Window Function,分别是ReduceFunction、AggreateFunction、FoldFunction、ProcessWindowFunction,其中FoldFunction将逐渐不再使用;四种类型有分为增量聚合操作(ReduceFunction、AggreateFunction、FoldFunction)和全量聚合操作(ProcessWindowFunction); 增量聚合函数计算性能高,占用存储空间少,因为其只需要维护窗口的中间结果状态值,不需要缓存原始数据;全量聚合函数使用代价相对高,性能较弱,因为算子需要缓存该窗口的接入数据,然后等窗口触发后对所有原始数据进行汇总计算,若接入数据量大或窗口时间长容易导致计算性能下降; ReduceFunction和AggreateFunction相似,但前者的输出类型和输入类型一致(如使用tuple的某个字段聚合),后者更加灵活地提供3个复写方法,add()定义数据的添加逻辑,getResult()定义根据Accumulator计算结果的逻辑,merge()方法定义合并accumulator的逻辑; ProcessWindowFunction可以支撑更复杂的算子,其支持基于窗口全部数据元素的结果计算,当算子需要窗口的元数据或状态数据,或者算子不支持运算交换律和结合律(统计所有元素的中位数和众数),需要该函数中的Context对象,Context类定义了Window的元数据及可以操作的Window的状态数据包括GlobalState和WindowState; 大部分情况下,需要增量计算和全量计算结合,因为增量计算虽然一定程度能够提升窗口性能,但灵活性不及ProcessWindowFunction,两者整合使用,既可以得到增量算子又可以得到窗口的元数据(窗口开始、终止时间等),比如在计算TOP N的场景中,分窗口计算完数据的计算后需要根据商品ID汇聚总的点击数; Watermark 由于网络或系统等外部因素影响,事件数据不能及时传输到Flink系统中,导致数据乱序、延迟等问题,因此需要一种机制能够控制数据处理的过程和进度;基于event_time时间的Windows创建后,具体如何确定属于该Windows中的数据元素已经全部到达,如果确定全部到达就可以对所有数据进行窗口计算操作(汇总、分组),如果数据没有全部到达,则继续等待该窗口中的数据,但是又不能无限期的等下去,需要有机制来保证一个特定的时间后,必须触发window去进行计算了,此时watermark发挥作用了,它表示当达到watermark后,在watermark之前的数据已经全部达到(即使后面还有延迟的数据);Watermark是处理EventTime 窗口计算提出的机制,本质上是一种时间戳,可以在读取 Source时候指定或者在transformation操作之前,用自定义的Watermark生成器按照需求指定; 正常情况下,流式数据的到达时间是有序的,如下图: 一般情况存在数据的乱序(out-of-order)和延迟(late element),此时水位线机制能表明该时间戳之前到当前水位线时间戳的数据已经全部达到,没有比它(水位线)更早的数据了,并触发计算; Flink中生成水位线的方式有两种:Periodic Watermarks(周期性)和Punctuated Watermarks,前者假设当前时间戳减去固定时间,所有数据都能达到,后者要在特定事件指示后触发生成水位线; 举例说明Periodic Watermarks 工作方式:当前window为10s,设想理想情况下消息都没有延迟,那么eventTime等于系统当前时间,假如设置watermark等于eventTime的时候,当watermark = 00:00:10的时候,就会触发w1的计算,这个时后因为消息都没有延迟,watermark之前的消息(00:00:00~00:00:10)都已经落入到window中,所以会计算window中全量的数据。那么假如有一条消息eventTime是00:00:01 应该属于w1,在00:00:11才到达,因为假设消息没有延迟,那么watermark等于当前时间,00:00:11,这个时候w1已经计算完毕,那么这条消息就会被丢弃,没有加入计算,这样就会出现问题。这是已经可以理解,代码中为什么要减去一个常量作为watermark,假设每次提取eventTime的时减去2s,那么当data1在00:00:11到达的时候,watermark是00:00:09这个时候,w1还没有触发计算,那么data1会被加入w1,这个时候计算完全没有问题,所以减去一个常量是为了对延时的消息进行容错; Punctuated Watermarks提供自定义条件生成水位,例如判断某个数据元素的当前状态或tuple类型的某个值,如果接入事件中状态为0则触发生成watermark,如果状态不为0则不触发,需要分别复写extractTimestamp和checkAndGetNextWatermark方法; Flink允许提前预定义数据的提取器Timestamp Extractors,在读取source时候定义提取时间戳; 延迟数据 基于Event_time的窗口计算虽然可以使用warterMark机制容忍部分延迟,但只能一定程度的缓解该问题,无法应对某些延迟特别严重的场景。Flink默认丢失延迟数据,但用户可以自定义延迟数据的处理方式,此时需要Allowed Lateness机制近数据的额外处理; DataStream API提供Allowed Lateness方法指定是否对迟到数据进行处理,参数是Time类型的时间间隔大小,代表允许的最大延迟时间,Flink的窗口计算中会将Window的Endtime加上该时间作为窗口最后释放的结束时间(P),当接入的数据中Event time未超过该时间(P),但WaterMark已经超过Window的Event_Time时直接触发窗口计算,若Event_Time超过了时间P,则做丢弃处理; 通常情况下可以使用sideOutputLateData 方法对迟到数据进行标记,然后使用getSideOutput()方法得到被标记的延迟数据,分析延迟原因; 多流合并/关联 合并 Connect:Flink 提供connect方法实现两个流或多个流的合并,合并后生成ConnectedStreams,会对两个流的数据应用不同的处理方法,并且双流之间可以共享状态(比如计数);ConnectedStream提供的map()和flatMap()需要定义CoMapFunction和CoFlatMapFunction分别处理输入的DataStream数据集; Union:Union算子主要实现两个或者多个输入流合并成一个数据集,需要保证两个流的格式一致,输出的流与输入完全一致; 关联 Flink支持窗口的多流关联,即在一个窗口上按照相同条件对多个输入流进行join操作,需要保证输入的Stream构建在相同的Windows上,且有相同类型的Key做为关联条件; 数据集inputStream1通过join方法形成JoinedStreams类型数据集,调用where()方法指定inputStream1数据集的key,调用equalTo()方法指定inputStream2对应关联的key,通过window()方法指定Window Assigner,最后通过apply()方法中传入用户自定义的JoinFunction或者FlatJoinFunction对输入数据元素进行窗口计算; Windows Join过程中所有的Join操作都是Inner Join类型,也就是必须满足相同窗口中,每个Stream都有Key,且key相同才能完成关联操作并输出结果; 状态和容错 有状态计算是Flink重要特性,其内部存储计算产生的中间结果并提供给后续的Function或算子使用,状态数据维系在本地存储中,可以是Flink的堆内存或者堆外内存中,也可以借助于第三方的存储介质,同storm+ redis / hbase模式相比,Flink完善的状态管理减少了对外部系统的依赖,减少维护成本; State和类型 Flink根据数据集是否根据key分区将状态分为Keyed State和 Operator State两种类型,Keyed State只能用于KeyedStream类型数据集对应的Function和Operation上,它是Operator State的特例; Operator State只和并行的算子实例绑定,和数据元素中的key无关,支持当算子实例并行度发生变化后自动重新分配状态数据; Keyed State和 Operator State均有两种形式,一种是托管状态,一种是原始状态,前者有Flink Runtime控制和管理状态数据并将状态数据转换成内存Hash tables 或RocksDB的对象存储,后者由算子自己管理数据结构,当触发CheckPoint后,Flink并不知道状态数据内部的数据结构,只是将数据转换成bytes数据存储在CheckPoint中,当从Checkpoint恢复任务时,算子自己反序列化出状态的数据结构; CheckPoint 和SavePoint Flink基于轻量级分布式快照算法提供了CheckPoint机制,分布式快照可以将同一时间点的Task/Operator状态数据全局统一快照处理,包括Keyed State和Operator State Savepoints是检查点的一种特殊实现,底层使用CheckPoint机制,Savepoint是用户以手工命令方式触发CheckPoint,并将结果持久化到指定的存储路径中,其主要目的是帮助用户在升级和维护集群过程中保存系统的状态数据,避免因停机运维或者升级到知道正常终止的应用数据状态无法恢复。 参考内容: https://www.cnblogs.com/leesf456/p/11136344.html https://blog.csdn.net/a_drjiaoda/article/details/89357916 https://www.jianshu.com/p/9e92cefa9d4e http://wuchong.me/blog/2018/11/18/flink-tips-watermarks-in-apache-flink-made-easy/ - END - 欢迎加入 大数据 |数仓技术交流群 。 进群方式:请加微信(微信号:dataclub_bigdata),回复:数据,通过审核会拉你进群。 (备注:行业-职位-城市) 福利时刻 01. 后台回复「数据」,即可领取大数据经典资料。 02.后台回复「转型」,即可传统数据仓库转型大数据必学资料。 03. 后台回复「加群」,或添加一哥微信ID:dataclub_bigdata拉您入群(大数据|数仓|分析)或领取资料。 !关注不迷路~ 各种福利、资源定期分享! 你点的每个 在看 ,我都认真当成了喜欢 本文分享自微信公众号 - 数据社(DataClub)。如有侵权,请联系 support@oschina.cn 删除。本文参与“OSC源创计划”,欢迎正在阅读的你也加入,一起分享。

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

STL笔记1

顺序容器有vector、list、deque。关联容器有map、set。容器类自动申请和释放内存,无需new和delete操作。但是需要连接STL各个容器的内存管理 STL六大组件:容器,算法,迭代器,仿函数、适配器和空间配置器容器:容纳一组元素的对象迭代器:提供一种访问容器中每一个元素的方法适配器:用来修饰容器,比如queue和stack,底层借助了deque。空间适配器:负责空间配置和管理 空间配置器:对象构造前的空间配置和对象析构后的空间释放,由负责。设计哲学如下:先system heap要求空间考虑多线程状态考虑内存不足时的应变措施考虑碎片问题 对于碎片问题,有双层及配置器:第一级直接使用allocate()调用malloc()、deallocate()调用free(),使用类似new_handler机制解决内存不足(抛出异常),配置无法满足的问题(如果在申请动态内存时找不到足够大的内存块,malloc 和new 将返回NULL 指针,宣告内存申请失败)。 第二级视情况使用不同的策略,当配置区块大于128bytes时,调用第一级配置器,当配置区块小于128bytes时,采用内存池的整理方式:配置器维护16个(128/8)自由链表,负责16种小型区块的此配置能力。内存池以malloc配置而得,如果内存不足转第一级配置器处理。 1、第一级配置器详解 2、第二级空间配置器详解 第二级空间配置器实际上是一个内存池,维护了16个自由链表。自由链表是一个指针数组,有点类似与hash桶,它的数组大小为16,每个数组元素代表所挂的区块大小,比如free list[0]代表下面挂的是8bytes的区块,free list[1]代表下面挂的是16bytes的区块…….依次类推,直到free _ list[15]代表下面挂的是128bytes的区块。 3、空间配置器存在的问题自由链表所挂区块都是8的整数倍,因此当我们需要非8倍数的区块,往往会导致浪费。 由于配置器的所有方法,成员都是静态的,那么他们就是存放在静态区。释放时机就是程序结束,这样子会导致自由链表一直占用内存,自己进程可以用,其他进程却用不了。 各种容器的特点和适用情况:vector:可变大小的数组,支持快速随机访问,在尾部之外的位置插入或删除元素会较慢。deque:双端队列,支持快速随机访问,在头尾位置插入删除速度快。list:双向链表,只支持双向顺序访问,在list任何位置插入/删除速度都很快。forward_list:单向链表,只支持单向顺序访问array:固定大小的数组支持随机访问,不能添加或者删除元素。string:和vector相似,随机访问快,在尾位置插入/删除元素快。

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

Scala Symbol笔记

Symbol This class provides a simple way to get unique objects for equal strings. Since symbols are interned, they can be compared using reference equality. symbols可以作为一种快速比较字符串的方式,如果字符串的值相同,则返回的symbol变量具有相同的引用地址。Symbol内部维护了一个字符串池。 object SymbolDemo { def main(args: Array[String]): Unit = { val s = 'nihao val n = 'nihao // return true println(s == n) } } Compare to Java 在Java中创建String实例有两种方式:1、直接给一个变量赋值;2、用new关键创建String对象;(下文记作:方式1 和 方式2) 方式1 我们都知道 “方式1” 每次都会创建一个新变量(所以for循环内拼接字符串不建议用 “+” 操作符,因为每次都会开辟一个新的内存)。但Java其实对该操作做了优化,在String类内部维护了一个字符串池,每次通过 “方式1” 创建String实例时,首先检查字符串池中有没有相同的字符串,如果字符串池中不存在该字符串,则将字符串放入字符串池中(此处开辟新内存),同时将字符串的引用地址赋值给变量;如果字符串池中存在该字符串,则直接将原有引用地址赋值给新变量。 当创建 “str1” 时,字符串池中没有 “Hello Str”,此时将 “Hello Str” 放入字符串池中,并将内存地址赋值给 “str1”。当创建 “str2”时,字符串池中已经存在 “Hello Str”,直接将原有内存地址赋值给 “str2”,所以 “str1 == str2” 返回 true 。 方式2 每次都会创建一个新的对象,当调用 intern() 时逻辑过程跟 “方式1” 相同。当字符串池中存在 “Hello Str”时直接返回内存地址,否则将 “Hello Str” 放入字符串池中,并返回内存地址。 public class Demo { public static void main(String[] args) { String str1 = "Hello Str"; String str2 = "Hello Str"; String str3 = new String("Hello Str"); // return true System.out.println(str1 == str2); // return false System.out.println(str1 == str3); // return true System.out.println(str1 == str3.intern()); } } Refer from: https://www.scala-lang.org/api/2.12.1/scala/Symbol.htmlhttps://stackoverflow.com/questions/3554362/purpose-of-scalas-symbol

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

leetCode 编程笔记

问:编写一个函数来查找字符串数组的最长公共前缀。如果不存在公共前缀,返回空字符串 “”。 示例输入:["flower", "flow", "flight"] 示例输出:"fl" 示例输入:["dog", "racecar", "car"] 示例输出:"" 解释:输入不存在公共前缀。 tips:所有的输入只包含小写字母 a-z 。 public class Solution { // 1. Method 1, start from the first one, compare prefix with next string, until end; // 2. Method 2, start from the first char, compare it with all string, and then the second char // I am using method 1 here public String longestCommonPrefix(String[] strs) { if (strs == null || strs.length == 0) { return ""; } String prefix = strs[0]; for(int i = 1; i < strs.length; i++) { int j = 0; while( j < strs[i].length() && j < prefix.length() && strs[i].charAt(j) == prefix.charAt(j)) { j++; } if( j == 0) { return ""; } prefix = prefix.substring(0, j); } return prefix; } } 有一个X x Y的网格,只能向右、向下移动,从(0, 0)走到(X - 1, Y - 1),中间某些位置有障碍物,打印一条路径(优化) 答:解题思路。 12.png 计算过程: 把底边和右边的每个格子标记为1 其余格子从右下角往右上角依次遍历 每个格子的值是其右边和下边格子值的和 遍历到右上角后求得最终结果 例如上图的结果为10。 这种解法依据的思路,由于每个格子只能向右或向下走,那么它的走法就由其右边格子的走法和下边格子的走法之和。而最下边和最右边的每个格子都只有唯一的走法。由此就能推导出其余格子的走法。 13.png tips:注意把握思想,把右下角右边和底部的边都设置为 1.之后向上推倒即可。最终结果是 10. 归并排序: //将有序数组a[]和b[]合并到c[]中 void MemeryArray(int a[], int n, int b[], int m, int c[]) { int i, j, k; i = j = k = 0; while (i < n && j < m) { if (a[i] < b[j]) c[k++] = a[i++]; else c[k++] = b[j++]; } // 当其中一个列表的所有数据都比另一个列表的所有数据小的时候,例如 i = n,j = 0; while (i < n) c[k++] = a[i++]; while (j < m) c[k++] = b[j++]; } tips: 解决了上面的合并有序数列问题,再来看归并排序,其的基本思路就是将数组分成二组A,B,如果这二组组内的数据都是有序的,那么就可以很方便的将这二组数据进行排序。如何让这二组组内数据有序了? 可以将A,B组各自再分成二组。依次类推,当分出来的小组只有一个数据时,可以认为这个小组组内已经达到了有序,然后再合并相邻的二个小组就可以了。这样通过先递归的分解数列,再合并数列就完成了归并排序。 构建 Hash 表的时候(散列函数的设计) f( key ) = key mod p ( p ≤ m ) mod ... 使用除留余数法的一个经验是,若散列表表长为m,通常p为小于或等于表 如何合理选取p值 使用除留余数法的一个经验是,若散列表表长为m,通常p为小于或等于表长(最好接近m)的最小质数或不包含小于20质因子的合数。 这句话怎么理解呢?要不这样吧,我再举个例子:某散列表的长度为100,散列函数H(k)=k%P,则P通常情况下最好选择哪个呢?A、91 B、93 C、97 D、99 实践证明,当P取小于哈希表长的最大质数时,产生的哈希函数较好。我选97,因为它是离长度值最近的最大质数。 可以盛最多水的容器 解法:我们可以循环遍历所有的两天边的乘积,取最大的值。 01.png 编程试题:求数列的和 使用语言:JAVA 参考正解代码如下: import java.util.*; class Main{ public static void main(String args[]){ int m; double sum,n; Scanner sc = new Scanner(System.in); while(sc.hasNext()){ n=sc.nextInt(); m=sc.nextInt(); sum=0; for(int i=0;i<m;i++){ sum=sum+n; n=Math.sqrt(n); } System.out.printf("%.2f",sum); System.out.println(); } } } 使用语言:C++ 参考正解代码如下: #include <math.h> #include <stdio.h> int main() { int n; double x, s; while (~scanf("%lf%d", &x, &n)) { for(s = 0.0; n--; x = sqrt(x)) s += x; printf("%.2lf\n", s); } return 0; } 使用语言:C# 参考正解代码如下: using System; namespace myApp { class Program { public static void Main() { string line; string[] p; int m, n; double nn; while (!string.IsNullOrEmpty(line = Console.ReadLine())) { p = line.Split(' '); n = int.Parse(p[0]); m = int.Parse(p[1]); double sum = 0; nn = n; for (int i = 0; i < m; i++) { sum = sum + nn; nn = Math.Sqrt(nn); } Console.WriteLine(string.Format("{0:f}", sum)); } } } } 使用语言:JavaScript 参考正解代码如下: var m; var sum,n; var sc while(sc = read_line()){ var arr = sc.split(' '); n=parseInt(arr[0]); m=parseInt(arr[1]); sum=0; for(var i=0;i<m;i++){ sum=sum+n; n=Math.sqrt(n); } print(sum.toFixed(2)); } 注意上面的三个代码有几点注意: 一:JavaScript 的输入和输出为 sc = read——line() 输出:print(); 精确到后两位:sum.toFixed(2) 二:C / C++ 精确到两位小数为:printf("%.2lf\n", s); 三:Java编写程序的时候精确到小数点后两位的写法:System.out.printf("%.2f",sum); 水仙花的求解 编程试题:水仙花 使用语言:JAVA 参考正解代码如下: import java.util.Scanner; public class Main{ public static void main(String args[]){ Scanner reader=new Scanner(System.in); while(reader.hasNextInt()){ int m=reader.nextInt(); int n=reader.nextInt(); if(100<=m&&m<=n&&n<=999){ int j=0; for(int i=m;i<=n;i++) { int geWei,shiWei,baiWei; baiWei=i/100; shiWei=(i-baiWei*100)/10; geWei=i-baiWei*100-shiWei*10; if(i==geWei*geWei*geWei+shiWei*shiWei*shiWei+baiWei*baiWei*baiWei) {j=j+1; if(j>1){ System.out.print(" "+i); } else{ System.out.print(i); } } } if(j==0){ System.out.print("no"); } System.out.println(); } } } } 使用语言:C++ 参考正解代码如下: #include<stdio.h> int main(){ int m,n; while(scanf("%d%d",&m,&n)!=EOF){ int t=0; for(int i=m; i<=n; i++){ int a=i/100; int b=i%100/10; int c=i%10; if(i==a*a*a+b*b*b+c*c*c && t==0){ printf("%d ",i); t++; } else if(i==a*a*a+b*b*b+c*c*c && t==1){ printf("%d ",i); } } if(t!=0){ printf("\n"); } if(t==0){ printf("no\n"); } } return 0; } 使用语言:C# 参考正解代码如下: using System; namespace myApp { class Program { public static void Main() { string line; string[] p; int m, n; while ((line = Console.ReadLine()) != null) { p = line.Split(' '); n = int.Parse(p[1]); m = int.Parse(p[0]); var j=0; for(var i=m;i<=n;i++) { int geWei,shiWei,baiWei; baiWei = (i/100); shiWei = ((i-baiWei*100)/10); geWei = i-baiWei*100-shiWei*10; if(i==geWei*geWei*geWei+shiWei*shiWei*shiWei+baiWei*baiWei*baiWei) { j=j+1; if(j>1) { Console.Write(" "+i); } else { Console.Write(i); } } } if(j==0) { Console.Write("no"); } Console.Write("\r\n"); } } } } 使用语言:JavaScript 参考正解代码如下: var sc; while(sc = read_line()){ var arr = sc.split(' '); n=parseInt(arr[1]); m=parseInt(arr[0]); if(100<=m&&m<=n&&n<=999){ var out = []; var j=0; for(var i=m;i<=n;i++) { var geWei,shiWei,baiWei; baiWei=parseInt(i/100); shiWei=parseInt((i-baiWei*100)/10); geWei=i-baiWei*100-shiWei*10; if(i==geWei*geWei*geWei+shiWei*shiWei*shiWei+baiWei*baiWei*baiWei) { j=j+1; if(j>1){ out.push(" "+i); } else{ out.push(i); } } } if(j==0){ out.push("no"); } print(out.join('')); } }

资源下载

更多资源
Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称,一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集,帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring

Spring框架(Spring Framework)是由Rod Johnson于2002年提出的开源Java企业级应用框架,旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念,提供核心容器、应用上下文、数据访问集成等模块,支持整合Hibernate、Struts等第三方框架,其适用范围不仅限于服务器端开发,绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux

Rocky Linux(中文名:洛基)是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版,作为CentOS稳定版停止维护后与RHEL(Red Hat Enterprise Linux)完全兼容的开源替代方案,由社区拥有并管理,支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性,采用模块化包装和SELinux安全架构,默认包含GNOME桌面环境及XFS文件系统,支持十年生命周期更新。

WebStorm

WebStorm

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

用户登录
用户注册