Posts categorized in ‘rakulang’ (533)

Roles as Algebraic Data Types in Raku

Raku and Rust Ffi

Grammar Composite

Empty

New Disp Is Here

Haku a Japanese Programming Language

105 C++ Algorithms in One Line of Raku

Feed Operator

Raku Shortcuts Part One

Raku Types

Comb in Raku

Roundrobin

可重播的 Process Functions

Actions and Ast in Raku

Accessing Data Structure Elements by Index or Key

Janes Fruit

List Comprehensions

Dynamic Variables in Raku

Terms in Raku

Signature in Raku

Raku Examples

Difference Between Rules and Tokens

Apropos proto: Perl6.c multi thoughts

A Mutable Grammar for Raku

Comb in Raku

Guess Who

List in Raku

Raku Blogs

Raku Grammars

Given When Pattern Match in Raku

Io in Raku

Polymod Method in Raku

Proto in Raku

Roles in Raku

The S Operator in Raku

An Example of Slip

Smile Type in Raku

Twigils in Raku

Unspace in Raku

Metaobject in Raku

One Liners in Raku

Variables in Raku

Quote in Raku

Introspection in Raku

Roles Conflict in Raku

Object Orientation in Raku

Terms in Raku

Raku Intro

Raku Calendar(2010)

Raku Core Hacking Grammatical Babble

The King of List Manipulation

Sneaky Methods

Protocol in Swift and Roles in Raku

Subscripts Adverbs

From Regex to Grammar Part One

An Example of Parsing Text in Grammar

Use Grammar to Parse Date

Add Subscripts for Custom Class in Raku

Add a Method to an Existing Class in Raku

Execute an External Program With Timeout in Raku

How to Write Custom Accessor Methods in Raku

There Is More Than One Way to Extend It

Find the Common Lines in Two Files

Exploring Raku Signatures Part Two

Hash Is Also Containers

Add Separator

Grammar Generating Grammar

Junction in Raku

Capture in Raku

Operator in Raku

Modules in Raku

Features in Raku

Raku Calendar 2012

Sneaking Into a Loop

Subset

When and Where

Raku Calendar

Roles or When One Is Many

Raku Syntax I Miss in Other Languages

Raku Multiple Dispatch With the New Moarvm Dispatcher

Raku Multiple Dispatch With the New Moarvm Dispatcher

Psql

Towards a New General Dispatch Mechanism in Moarvm

assumed predictability

making myself a present

coercive files

on coercion method return value

The Strange Case of the Greedy Junction

Reconstructing Raku’s Junctions

先说点题外话。在写这篇文章的过程中，我遭遇了系统管理员最可怕的噩梦：服务器丢失，然后是备份不好。直到最后一刻，我都有充分的理由担心自己根本无法完成这篇文章。幸运的是，我与生命做了休战，得到了暂时的喘息。结论是什么？不要在 ESXi 上使用 bareos。或者，可能就是不要使用裸 OS… 在为我之前的 advent 日志领取 RFC 的时候，我完全把注意力放在了语言对象部分。我花了几遍时间才找到合适的内容。但在这期间，我发现列表中居然少了一个非常重要的主题。“不可能！” - 我自言自语道，随后又继续搜索。然而，无论是搜索"抽象类"，还是搜索"角色"，都没有任何结果。我本想放弃，并做出结论，这个想法是后来才有的，当时写提纲的时候还是这样的。 但是，等等，OO 相关的 RFC 中提到的主题是什么接口？哦，那个接口! 就像请求正文中所说的那样。 增加一个机制来声明类的接口 还有一个方法来声明一个类实现了上述接口。 此时我再次意识到，现在已经落后了整整20年。这段文字来自于很多人认为 Java 是唯一正确的 OO 实现的时代! 而事实上，通过进一步的阅读，我们发现下面的说法，很可能是受到当时一些流行观点的影响。 现在，如果一个接口文件试图做任何事情，而不是预先声明方法，这都是一个编译时错误。 这让人想起了什么，不是吗？然后，在 RFC 的最后，我们又发现了一个问题。 Java 是一种使用接口多态性的语言。不要被它吓倒–如果我们一定要从 Java 中偷点东西，那就偷点好东西吧。 好东西？好东西？哦，我的……Java 试图通过简单地否认 C++ 多重继承方式来解决它的问题，这是让我从一开始就远离这门语言的原因。早在90年代初，我就受够了 Pascal 控制我的写作风格! 幸运的是，那些参与早期 Perl6 设计的人一定和我一样对这个问题有相同的看法（此外，Java 本身也发生了很大的变化）。所以，我们现在有了角色。它们与抽象类和现代接口的共同点是，一个角色可以定义一个接口来与一个类进行通信，也可以提供一些特定角色行为的实现。它还可以做得更多一些，不仅仅是这些! 角色的不同之处在于 Raku OO 模型中角色的使用方式。一个类不会实现一个角色；也不会像抽象类那样从它那里继承。相反，它做的是角色；或者我喜欢用另一个词来形容：它消耗一个角色。从技术上讲，这意味着角色被混入类中。这个过程可以形象地描述为：如果编译器将角色的类型对象所包含的所有方法和属性，重新植入到类中。类似这样。 role Foo { has $.foo = 42; method bar { say "hello!" } } class Bar does Foo { } my $obj = Bar.

Perl Should Stay Perl

Pair Constructor

2000年9月7日提出，2000年9月20日冻结，取决于 RFC 159: True Polymorphic Objects 2000年8月25日提出，2000年9月16日冻结，也是由Nathan Wiger提出的，之前已经在博客中提到过。 tie 到底是什么？ RFC 200 是关于扩展 Perl 提供的 tie 函数。 Perl 中的这个功能允许人们将程序逻辑注入到系统对标量、数组和哈希等的处理中。这是通过给数组等数据结构分配一个包的名称来实现的（也就是绑定）。然后，该包要提供一些子程序（如 FETCH 和 STORE），这些子程序将被系统调用，以实现对给定数据结构的某些效果。 因此，它被 Perl 的一些核心模块（如线程）和 CPAN 上的许多模块（如 Tie::File）所使用。Perl 的 tie 函数仍然存在 RFC 中提到的问题。 都是绑定的 在 Raku 中，所有的东西都是一个对象，或者可以被认为是一个对象。系统需要对一个对象做的一切事情，都是通过它的方法来完成的。在这个意义上，你可以说，Raku 中的所有东西都是一个绑定的对象。幸运的是，Rakudo（Raku 编程语言最先进的实现）可以识别出对象上的某些方法实际上是系统提供的方法，并在编译时实际创建捷径（例如，当分配给一个有标准容器的变量时：它实际上不会调用 STORE 方法，而是使用一个内部子程序来达到预期的效果）。 但除此之外，Rakudo 还具有在程序执行过程中识别热点代码路径的能力，并对这些路径进行实时优化。 Jonathan Worthington 就这个过程做了两场非常精彩的演讲。从2017年的去优化如何帮助我们更快，以及2019年的性能更新。 因为在 Raku 中所有的东西都是一个对象，并且通过这些对象的类的方法进行访问，这使得编译器和运行时能够更好地掌握程序中实际发生的事情。从而获得更好的优化能力，甚至在某些时候优化到机器语言级别。 因为在 Raku 中一切都被 “绑定"了（用 Perl 过滤过的眼镜来看），将程序逻辑注入到系统对数组和哈希的处理中，可以简单到只需子类化系统的类，并提供一个系统使用的标准方法的特殊版本。假设你想在你的程序中看到当一个元素从一个数组中获取时，只需要添加一个自定义的 AT-POS 方法: class VerboseFetcher is Array { # subclass core's Array class method AT-POS($pos) { # method for fetching an element say "fetching #$pos"; # tell the world nextsame # provide standard functionality } } my @a is VerboseFetcher = 1,2,3; # mark as special and initialize say @a[1]; # fetching #1␤2 Raku 文档中包含了一个概述，说明要模拟一个 Array 和模拟一个 Hash 需要提供哪些方法。顺便说一下，关于通过索引或键来访问数据结构元素的整个词条是推荐给想要了解 Raku 内部的这些方面的人阅读的。

true polymorphic objects

Failed to find dependencies

peaked trim with raku

brace matching

assignment within regex

multiline comments

多路比较

我一直在家里静静地玩着每周挑战，本周的第一个任务是: 写一个脚本，找到第一个至少有5个不同数字的平方数。 解决这个问题的方法是（很明显！）懒洋洋地把每一个数字从1到无穷大的平方，然后梳理每个平方的数字，寻找5个或更多独特的数字，并立即输出你找到的第一个这样的平方。 这就直接翻译成 Raku: 1..∞ ==> map {$^n²} ==> first {.comb.unique ≥ 5} ==> say(); 但那个解决方案的优雅并不是我喜欢 Raku 的原因。 我爱 Raku 是因为，如果那个解决方案对你来说太可怕了（太无限、太懒惰、太并发、太管道化、太 Unicode、太声明式、太函数化、太像 Erlang 大师会写的代码），那么 Raku 同样允许你写一个简单的版本：一个命令式的、迭代的、块结构的、变量驱动的、纯 ASCII 的、或多或少完全和你在 Perl 中写的一样的版本，甚至在C中： loop (my $n=1 ;; $n++) { my $n_squared = $n ** 2; my %unique-digits; for (split '', $n_squared, :skip-empty) { %unique-digits{$_}++ } if (%unique-digits >= 5) { say $n_squared; last; } } 或者你也可以很容易地写出一个介于这两个极端之间的解决方案，无论其复杂程度和分解程度如何，都是你个人舒适区的甜蜜点。比如说： sub find_special_square { for 1..Inf -> $n { return $n² if $n².

上周每周挑战的第二个任务是列出某年每个月最后一个星期五的日期。 许多参与者花了很大的力气来创建高效而准确的解决方案：跟踪每个月的最后一天，正确检测闰年，计算出特定日期的月日公式，管理所需的复杂日期运算。 但是，鉴于 Raku 中内置了强大的 Date 类，这些令人钦佩的劳作其实大部分都没有必要。 整个任务可以通过简单地从给定年份的1月1日到12月31日来完成，首先检查每个日期是否是星期五（即"星期几"的值是5），然后检查下一个星期五（即正好一周后的日期）是否恰好在不同的月份。 如果这两个条件都为真，那么我们就有一个"本月最后一个星期五"，所以我们只需打印出来。 换句话说: for Date.new($year,1,1) .. Date.new($year,12,31) -> $date { if $date.day-of-week == 5 && $date.later(:1week).month != $date.month { say $date; } } 我们不需要明确地担心每个月有多长，一年是否是闰年，或者如何计算出某一天是否是星期五；Date 类为我们解决了所有这些问题。 而这不是运气，完全是设计好的。Raku 有大量类似这样的内置数据类型；它们是另一种"合适的工具，就在手边"。 当然，因为 Raku 是多范式的，所以有很多其他的方法可以将解决方案包裹在 Date 类的便利性中，这取决于你喜欢如何思考这个世界。 如果你喜欢纯韩式式的解决方案，可以这样写: say join "\n", grep { .later(:1week).month != .month }, grep { .day-of-week == 5 }, Date.new($year,1,1) .. Date.new($year,12,31); 或者你更喜欢一个（潜在的）并行的管道，然后通过一系列独立的过滤器和处理器来过滤日期序列: Date.new($year,1,1) .. Date.new($year,12,31) ==> grep( {.day-of-week == 5} ) ==> grep( {.

叹息。它总是这样，不是吗？ 你很快就写完了关于拥有正确的工具如何使某一特定的编码任务变得微不足道…当你意识到，就在隔壁，有一个更好的例子，你可以用它来表达同样的观点。 我在上一篇文章中谈到的"最长初始子路径"的例子是上周每周挑战的第2项挑战。但那周的挑战1让我更清楚地看到，正确的工具可以简化一项任务。 挑战1是找到最小的不是质数的欧几里得数。第 N 个欧几里得数是由前 N 个质数的乘积加上1给出的。所以欧几里得数的序列是: (2)+1, (2*3)+1, (2*3*5)+1, (2*3*5*7)+1, (2*3*5*7*11)+1, ... 如果有一个简单的方法来建立另一个列表的部分乘积的序列（例如质数的部分乘积），这个序列就很容易计算了: (2), (2*3), (2*3*5), (2*3*5*7), (2*3*5*7*11), ... 当然，Raku 也有这样一个工具内置在核心语言中。它叫做三角还原元操作符。 在 Raku 中，一个普通的还原运算符(如 [*]): $product = [*] @nums; 在列表的每两个元素之间插入方括号内的操作符，就像已经写好的一样: $product = @nums[0] * @nums[1] * @nums[2] * ... * @nums[*-1]; 在这个例子中，这计算的是所有数字的乘积。 另一方面，三角还原元运算符(例如，“三角积"运算符：[/*]): @products = [\*] @n; 或多或少地做了同样的事情，但它不是只返回最后的计算结果，而是从每一个连续的乘法步骤中返回一个部分结果的渐进列表，就像它被写的那样: @products = (@n[0], @n[0]*@n[1], @n[0]*@n[1]*@n[2], ...); 这似乎是一个神秘而无用的功能，但是，当然，这正是我们实际需要的神秘而无用的功能，以便从质数列表中建立欧几里得数的列表。 记住，我们要找的是第一个非质数欧几里得数。也就是：从所有大于从2到无穷大的所有质数的累计部分乘积序列的数中，选出第一个非质数。 而这一描述，又一次直接转化为一行 Raku: say first !*.is-prime, # Print the first non-prime from map *+1, # all numbers one greater than [\*] # the cumulative products of grep &is-prime, # all the prime numbers 2.

几年前，我在 Raku 中创建了一个关于 “transparadigm 编程"的演讲（以及后来的整个课程）。 基本的前提是，当一些语言限制你只能使用一个单一的锤子（或者更糟的是：一个装满锤子的盒子）时，Raku 被设计成一个完整的工具箱：整合了 OO、函数式、并发、声明式和过程式工具，使你能够为每个工作选择正确的组合。 最近，这个想法又在我脑海中全面浮现。在上周的每周挑战中，第二个任务是把文件路径列表，找到最长的共同初始子路径（即它们都共享的最深的目录）。 各个注册参与者提供的解决方案都非常简洁，而且通常既高效又优雅。然而他们中的大多数都是同一个程序解决方案的变体。在目录分隔符上拆分每条路径，然后对于1到N： 比较所有第N个组件，如果它们不一样就退出。类似于: my @components = @list.map({.split('/')}; for 1..* -> $n { next if all(@components).elems > $n && [~~] @components.values».[$n]; say @components.first.[0..$n-1].join('/'); last; } 并不是说这种方法有什么不好。 只是感觉比实际需要的"体力劳动"多了很多；很多低级的程序化的"告诉我怎么做”，而不是高级的声明性或功能性的"告诉我做什么"。 当我自己来解决这个问题的时候，我是这样想的。 我的搜索空间是所有可能的初始子路径的所有路径。在这个空间内，我需要找到所有路径共享的最长的初始子路径。换句话说，我需要将每条路径转换成一组越来越长的子路径，然后找到这些集合的交集，再找到该交集中最长的元素。它可以直接翻译成 Raku，比如这样: @list.map({m:ex{^.*\/}».Str}).reduce({$^a∩$^b}).keys.max(*.chars).say; 或（对于喜欢注释的人）喜欢这样: @list\ # In the list... .map({m:ex{^ .* '/'}\ # Find all initial subpaths ».Str})\ # ...as lists of strings .reduce({$^a ∩ $^b})\ # Then find all shared subpaths .keys.max(*.chars)\ # Then find the longest .

第15周挑战赛的第二个任务是实现 Vigenère 密码的编码器和解码器。但这比看起来要复杂一些，因为以 Blaise de Vigenère 命名的密码其实并不是他发明的，而 Vigenère 真正发明的密码也不是以他的名字命名的。 所以我们应该实现维根埃密码…还是维根埃的密码？为什么不两者兼而有之呢! Vigenère 密码是由 Giovan Battista Bellaso 在1553年设计的，然后在大约三百年后被误认为是 Vigenère 发明的。它使用了一个 tabula rēcta 来将信息文本翻译成密码文本，然后再翻译回来。 给定一个用户提供的密钥（如 “BELLASO”），我们通过匹配密钥和信息中各自的字母来加密信息（如 “VIGENEREDIDNOTINVENTTHIS”），然后用它们作为两个索引，在 tabula rēcta 的相应列和行中查找相应的密文字符。而如果密钥比信息短，我们只需回收密钥，次数不限。 比如说: Key: B E L L A S O B E L L A S O B E L L A S O B E L L A... Text: V I G E N E R E D I D N O T I N V E N T T H I S Table: ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ Cipher: W M R P N W F F H T O N G H J R G P N L H I M D 换句话说，tabula rēcta 的每一列都是一个独立的凯撒密码（或 ROT-N 转录），钥匙的每一个连续字母都会选择对信息中的相应字母进行哪种替换。

上周每周挑战的第一个任务是打印前十个强质数和弱质数。如果一个质数 pn 大于其两个相邻质数的平均数（即 pₙ > (pₙ₋₁+pₙ₊₁)/2），则该质数为"强"质数。如果一个质数小于其两个相邻质数的平均值，那么它就是"弱"质数。 当然，如果我们碰巧有一个所有质数的列表，这个挑战就变得微不足道了。那么我们只需要过滤掉前十个强数，和前十个弱数。事实上，如果我们碰巧有一个强质数的列表和一个弱质数的列表，那就更简单了。那我们就只需要打印出每个的前十个。 但是，质数和弱质数的数量是无限的（也有可能是强质数，虽然这还只是猜测），所以在大多数编程语言中建立一个完整的质数亚种列表是不切实际的。 然而，我喜欢 Raku 的另一个原因是它能很好地处理无限列表和序列。不用担心有限的上限，简化了大量的任务…这就是其中之一。 质数序列只是正整数序列，经过过滤（用 .grep）只保留那些质数。当然，Raku 已经有一个质数测试器：内置的 &is-prime 函数。质数序列永远不会改变，所以我们可以把它声明为一个常数: constant p = [ (1..∞).grep( &is-prime ) ]; 现在我们只需要提取强质数和弱质数。鉴于前面的"强"和"弱"的定义，我们得到： sub strong (\n) { n > 0 && p[n] > p[n-1, n+1].sum/2 } sub weak (\n) { n > 0 && p[n] < p[n-1, n+1].sum/2 } 请注意，p[n-1，n+1] 是 p[n] 的两个邻域的列表，然后我们把它们加在一起(.sum)和平均(/2)。 这两个强度测试是在质数的索引上操作的，而不是它的值，所以要生成强质数和弱质数的完整列表，我们需要取这些索引（p.keys），过滤它们，只保留强或弱的索引（另一个 .grep），然后将每个索引映射回对应的质数。 这就得到了： constant strong-p = [ p.keys.grep(&strong).map(->\n { p[n] }) ]; constant weak-p = [ p.

上周每周挑战的第一个任务是生成范埃克序列: 0, 0, 1, 0, 2, 0, 2, 2, 1, 6, 0, 5, 0, 2, 6, 5, 4, 0,... 第一个挑战是要理解什么是范埃克序列，因为网上的各种解释都没有即时的帮助。 范埃克序列是一个从零开始的整数列表，序列中的下一个数字由当前数字与该数字最近的前一个出现的距离给出。 例如，如果当前指数N处的数字（我们称它为：Aₙ）是7，那么为了计算指数N+1处的数字，我们通过序列回溯7的最近一次出现（在某个较早的指数M处）。那么序列中的下一个数字就是这两个7的出现之间的距离: N - M 唯一复杂的是，如果当前数字在前面的序列中没有出现，那么下一个数字就是零。这并不像你想象的那样武断：我们可以把这个零值看作是 Aₙ 到自身的距离。也就是说，如果没有前面的出现，那么"前面"出现的唯一可能的候选者只是 Aₙ 本身，所以这种情况下的距离是 N - N，也就是零。 好了，这就是范埃克序列。现在我们如何在 Raku 中生成它？ 如果你一直在关注我最近的博客文章，那么了解到 Raku 有一个适当的内置工具，它将让我们通过一个单一的语句来完成整个任务，可能不会有太多的惊喜。这个工具就是序列运算符。 ... 序列操作符接受一个初始值列表，然后是一个代码对象（即一个块或子例程），接着是一个终止条件。 my @sequence = @initial-list, &code-obj ... $termination; 它从初始值开始建立一个序列（字面意思是 Seq 类的对象），然后通过在初始列表的最后元素上反复调用代码对象，在初始列表之后生成额外的值。每次调用代码对象时，它返回的值都会被追加到序列中，直到返回一个与终止条件智能匹配的值，这时序列就完成了。例如 # initial code-obj termination my @odd-nums = 1, 3, 5, {$^n + 2} ... 999; my @N-powers = $N, $N², {$^x * $N} .

HBase 表 create 'stream_kafka_offsets', {NAME=>'offsets', TTL=>1209600} RowKey 布局: row: <topicName>:<groupID>:<EPOCH_BATCHTIME_MS> column family: offsets qualifier: <PARTITION_ID> value: <OFFSET_ID> /* Save offsets for each batch into HBase */ def saveOffsets(topicName: String, groupID: String, offsetRanges: Array[OffsetRange], hbaseTableName: String, batchTime: org.apache.spark.streaming.Time) = { val hbaseConf = HBaseConfiguration.create() hbaseConf.addResource("src/main/resources/hbase-site.xml") val conn = ConnectionFactory.createConnection(hbaseConf) val table = conn.getTable(TableName.valueOf(hbaseTableName)) val rowKey = topicName + ":" + groupID + ":" + String.valueOf(batchTime.milliseconds) val put = new Put(rowKey.getBytes) for(offset <- offsetRanges){ put.

使用 Spark Structured Streaming 消费 Kafka 数据并实时保存为 parquet 文件时, 出现一个问题, 代码如下: val res: StreamingQuery = adapterData .repartition(sparkConf.numOfPartitions) .writeStream .format("parquet") .option("path", "/parquet_data/enterprise/") .partitionBy("vintype", "dt") .outputMode(OutputMode.Append()) .trigger(Trigger.ProcessingTime("1200 seconds")) .queryName("enterprise kafka data saving as parquet") .start() 问题： 在线上运行一段时间后, 发现 /parquet_data/enterprise/ 目录下不再有新的文件追加了, 查日志没发现原因。隔一段时间就出现一次。临时解决办法是删除并重建 _spark_metadata 目录和 checkpoint 目录。 这样会丢数据。

首先创建项目框架 在终端中执行如下命令: curl https://flink.apache.org/q/quickstart-scala-SNAPSHOT.sh | bash -s 1.10.0 此例子中使用了 flink-table 依赖, 我们在 pom 文件中加入它: <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table</artifactId> <version>${flink.version}</version> <type>pom</type> <scope>provided</scope> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-common --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-common</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency>

免责声明：是的，我知道这个话题有些争议，而且我知道大多数信息都在 Spark 文档中针对其 Streaming API 进行了介绍，但是当我看到这个错误发生了很多遍之后，我感到写这篇博客的冲动很强烈。 我经常会看到来自 Spark Streaming 的新手在 StackOverflow 上提问，大致如下所示： 问题：“我正在尝试执行 XYZ，但无法正常工作，该怎么办？ 这是我的代码：” val sparkContext = new SparkContext("MyApp") val streamingContext = new StreamingContext(sparkContext, Seconds(4)) val dataStream = streamingContext.socketTextStream("127.0.0.1", 1337) dataStream.foreachRDD { rdd => // Process RDD Here } 嗯，好吧，这是怎么了？ 当我开始学习 Spark 时，我的第一个着陆点是有关 RDD（弹性分布式数据集）如何工作的解释。通常的例子是单词统计，其中所有操作都在 RDD 上执行。我认为可以肯定的是，这是许多其他学习 Spark 的人的切入点（尽管如今 DataFrame\Sets 已成为初学者的首选方法）。 当人们飞跃使用 Spark Streaming 时，可能有点不清楚 DStream 的附加抽象意味着什么。这导致许多人寻求他们可以掌握的东西，他们遇到的最熟悉的方法是 foreachRDD，该方法将 RDD 作为输入并产生 Unit（典型的副作用方法的结果）。然后，他们可以再次在他们已经感到满意和理解的 RDD 级别上工作。这完全遗漏了 DStreams 的要点，这就是为什么我想简要了解一下我们可以在 DStream 本身上做些什么而无需探究底层 RDD 的原因。

更新(2017.08.01): Spark v2.2 最近推出了一个叫做 mapGroupsWithState 的有状态流的新抽象, 我最近有一篇博客也是关于它的。我强烈建议你检查一下。 介绍 Apache Spark 由几个模块组成，每个模块都有不同的用途。 它的功能强大的模块之一是 Streaming API，它使开发人员能够在称为 Discretized Stream 或 DStream 的抽象下使用连续流（或准确地说是微批次）。 在这篇文章中，我将深入探讨 Spark Streaming 的一个特殊属性，它是有状态的 Streaming API。 有状态流使我们能够维护微批之间的状态，从而使我们能够形成数据的会话化。 免责声明-为了遵循本文的流程，应该对 Spark 的工作原理有基本的了解，并对 DStream 抽象具有一般的了解。 如果没有，请继续阅读，不用担心，我会等你… 欢迎回来！ 让我们继续。 通过例子理解 为了了解如何使用 API​​，让我们创建一个简单的传入数据示例，该示例要求我们进行会话化。我们的输入数据流将是 UserEvent 类型： case class UserEvent(id: Int, data: String, isLast: Boolean) 每个事件描述一个唯一的用户。我们通过用户 ID 标识用户，并用 String 表示发生的事件的内容。我们还想知道用户何时结束会话，因此我们提供了一个 isLast 标志来指示会话结束。 我们负责汇总所有用户事件的状态将是 UserSession 类型的状态： case class UserSession(userEvents: Seq[UserEvent]) 其中包含特定用户发生的事件序列。在此示例中，我们假设数据源是来自于 Kafka 使用的 JSON 编码数据流。 我们的 Id 属性将用作键，而 UserEvent 将是我们的值。两个放在一块，我们得到一个 DStream[(Int, UserEvent)]。

所以，您想学习 Rust，并不断了解所有权和借用的概念，但不能完全了解它的含义。所有权至关重要，因此在学习 Rust 的过程中尽早理解它是很好的，而且还可以避免遇到导致您无法实现程序的编译器错误。 在上一篇文章中，我们已经从 JavaScript 开发人员的角度讨论了所有权模型。在本文中，我们将仔细研究 Rust 如何管理内存，以及为什么这最终会影响我们在 Rust 中编写代码并保持内存安全的方式。 那么什么是内存安全？ 首先，最重要的是要了解在讨论什么使 Rust 成为一种编程语言时，内存安全实际上意味着什么。特别是当来自非系统编程背景，或者主要具有垃圾回收语言的经验时，可能很难理解 Rust 的这一基本功能。 正如威尔·克里顿（Will Crichton）在他的伟大文章《Rust 的内存安全：C 语言案例研究》中所述： “内存安全性是程序的属性，其中所使用的内存指针始终指向有效内存，即已分配的内存和正确的类型/大小。内存安全是一个正确性问题-内存不安全程序可能会崩溃，或者会由于错误而产生不确定的输出。” 实际上，这意味着存在允许我们编写“内存不安全”代码的语言，从某种意义上来说，引入错误非常容易。其中一些错误是： 悬空指针：指向无效数据的指针（一旦我们查看数据在内存中的存储方式，这将更有意义）。您可以在此处阅读有关悬空指针的更多信息。 两次释放：尝试两次释放相同的内存位置，这可能导致“未定义的行为”。在这里查看更多信息。 为了说明悬空指针的概念，让我们看一下下面的 C++ 代码及其在内存中的表示方式： std::string s = "Have a nice day"; 初始化的字符串在内存中通常使用如下的栈和堆表示： buffer / capacity / / length / / / +–––+––––+––––+ stack frame │ • │ 16 │ 15 │ <– s +–│–+––––+––––+ │ [–│––––––––––––––––––––––––– capacity ––––––––––––––––––––––––––] │ +–V–+–––+–––+–––+–––+–––+–––+–––+–––+–––+–––+–––+–––+–––+–––+–––+ heap │ H │ a │ v │ e │ │ a │ │ n │ i │ c │ e │ │ d │ a │ y │ │ +–––+–––+–––+–––+–––+–––+–––+–––+–––+–––+–––+–––+–––+–––+–––+–––+ [––––––––––––––––––––––––– length ––––––––––––––––––––––––––] 我们将在一秒钟之内介绍一下栈和堆的内容，但是现在很重要的一点是，要知道存储在栈中的是 std::string 对象本身，它的长度为三个字，固定大小。这些字段是指向堆分配缓冲区的指针，该缓冲区保存实际数据，缓冲区容量和文本长度。换句话说，std::string 拥有其缓冲区。程序销毁该字符串时，也会通过该字符串的析构函数释放相应的缓冲区。

我有个 Spark FileStreaming 程序，当监控到一个批次完成后，就执行 hdfs 的 mv 命令： import scala.sys.process._ import org.apache.spark.streaming.StreamingContext import org.apache.spark.streaming.scheduler.{StreamingListener, StreamingListenerBatchCompleted} class StreamingMonitor(ssc: StreamingContext) extends StreamingListener { override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = { try { val day = Seq("sh", "-c", "hdfs dfs -cat /tmp/apps/days.txt | head -1").!!.trim // 取日期文本文件的第一行 Seq("sh", "-c", s""" if hdfs dfs -test -e /a26_adapter_data/nation/vintype=A26/d=${day}; then hdfs dfs -mv /a26_adapter_data/nation/vintype=A26/d=${day} /daily_parquet/nation/vintype=A26 fi """).!! Seq("sh", "-c", s""" if hdfs dfs -test -e /a26_adapter_data/enterprise/vintype=A26/dt=${day}; then hdfs dfs -mv /a26_adapter_data/enterprise/vintype=A26/dt=${day} /daily_parquet/enterprise/vintype=A26 fi """).

最近几周，我专注于 Apache Beam 项目。经过一番阅读之后，我发现了 Beam 和 Spark 结构化流之间的许多相似概念（或反之？）。相似之处之一就是触发器。 经过数月的休息后，本文介绍了另一个 Apache Spark 功能，即触发器。第一部分介绍了 Apache Spark 结构化流项目的上下文中的触发器。第二个显示了一些实现细节。最后一部分包含一些学习测试，显示了触发器如何工作。 触发器角色 Spark 触发器与 Apache Beam 中的触发器具有相似的作用，即，它们确定何时开始对累积数据进行处理。该处理的执行显然会向结果表发出新数据。关于 Apache Spark 中基于流的先前版本（基于 DStream），触发器的概念类似于批处理间隔属性。 在描述的版本（2.2.1）中，Spark 中有2种不同的触发器类型。第一种类型基于处理时间。它根据处理时间按固定间隔执行流查询。可以以任何单位时间（ms，s，min，…）定义此间隔。第二种类型称为once，因为它仅执行一次查询。执行后，查询终止，即使有新数据到达，查询也不会再次开始。 默认情况下，Apache Spark 结构化流以 0 ms的基于处理时间的触发器执行查询。这意味着 Spark 将在处理完之前的查询后尽快启动新查询。仅当存在新数据时，才会执行新的执行。 触发器内部 在内部，触发器分组在 org.apache.spark.sql.streaming.Trigger 类中，其中每种触发器类型都由一个或多个工厂方法表示。在处理时间的情况下，我们可以使用以下方法创建触发器：ProcessingTime（长间隔Ms），ProcessingTime（长间隔，TimeUnit timeUnit），ProcessingTime（持续时间间隔）或 ProcessingTime（字符串间隔）。它们全部都从 org.apache.spark.sql.streaming.ProcessingTime 对象调用幕后创建或应用方法。一次触发器由返回 OneTimeTrigger 案例对象的 Once() 表示。 如此创建的触发器实例稍后将在流查询中用作 org.apache.spark.sql.execution.streaming.StreamExecution 属性的一部分。在此实例中，触发器用于构建org.apache.spark.sql.execution.streaming.TriggerExecutor 实现的正确实例，该实现将是 ProcessingTimeExecutor 来处理基于时间的触发器，或者是 OneTimeExecutor 来执行一次触发器。 稍后，流查询由 TriggerExecutor的execute(triggerHandler: ()=> Boolean) 方法执行。此方法的实现取决于触发器类型。对于一次执行的触发器，execute 方法仅启动一次 TriggerHandler 函数。对于ProcessingTimeExecutor，execute 方法是一个长时间运行的过程（while(true)循环），其中触发器在执行查询之前等待间隔时间。 触发器还与 org.apache.spark.sql.execution.streaming.ProgressReporter#finishTrigger(hasNewData: Boolean) 方法中定义的统计信息有关。 触发器范例 下面的学习测试显示了一些触发特性： "once trigger" should "execute the query only once" in { val inputStream = new MemoryStream[Int](1, sparkSession.

结构化流引入了许多有关基于 DStream 的流的新概念。其中之一是输出模式。 这篇文章介绍了 Spark 2.0.0 中引入的用于处理流数据输出的输出模式。第一部分通过简短的理论部分向他们展示。第二部分介绍了它们的 API。最后一部分显示了它们在某些学习测试中的工作方式。 输出模式定义 输出模式指定将数据写入结果表的方式。在可用的输出模式中，我们可以区分: 追加(append)-这里仅将新行写入输出接收器。此模式保留给处理，不进行任何汇总，非常适合不可变的结果。 需要注意的重要一点是与水位的关系。如果在聚合中定义了水位，则它控制何时将附加数据发送到结果表。它仅在“中间”状态（=处理期间更改的状态）完成之后才发生。仅当水位的新值在给定组中的最新条目之前通过时，才会发生这种情况。例如，假设水位设置为19:00:00，并且组“A”的记录的最新值来自 19:00:05。现在，当水位传递到 19:00:06时，组“A”的结果将被发送到结果表。引擎认为不会传递任何新的事件，因此可以安全地将结果发送到输出表。 完成(compelete)-在这种模式下，所有行每次都被写入输出接收器。当流有一些更新时，将进行写入。此模式专用于具有聚合的处理。 更新(update)-与上一个相似，不同之处在于仅将更新的行写入输出接收器。 输出模式 API 输出模式定义发生在 DataStreamWriter#outputMode（outputMode: String）方法中。因此，传递的名称稍后会从 InternalOutputModes 对象转换为相应的 case 对象。 解析的实例主要在 DataStreamWriter 类中使用。它从那里传递到 StreamingQueryManager#startQuery(userSpecifiedName: Option[String], userSpecifiedCheckpointLocation: Option[String], df: DataFrame, sink: Sink, outputMode: OutputMode, useTempCheckpointLocation: Boolean = false, recoverFromCheckpointLocation: Boolean = true, trigger: Trigger = ProcessingTime(0), triggerClock: Clock = new SystemClock()).。顾名思义，此方法开始执行流查询。 在物理执行方面，我们可以在 StateStoreSaveExec 中找到输出模式的轨迹。在那里存储中间状态结果。顺便说一下，我们可以找到许多有关水位的参考，有助于消除过时的结果。如果您想了解更多信息，请转到 Apache Spark 结构化流中有关 StateStore 的文章。 输出模式示例 下表总结了可用于给定类型的处理的模式。在每个测试之后，编写一些测试以显示使用情况和未使用情况: 带水位的聚合 附加模式: 受支持-但结果仅在超过水位后才发出（=输出完成）。如果未将汇总应用于水位的列，则该查询将不起作用。 完整模式: 受支持-与更新不同，不使用水位 更新模式: 支持-水位用于删除太旧的聚合 测试:

结构化流可通过应用于状态管理，数据源和数据接收器的语义来保证端到端的一次精确交付（以微批处理模式）。关于状态的帖子更详细地介绍了状态，但还有2个其他部分尚待发现。 这篇文章分为2个主要部分。第一部分着眼于数据源，并说明了在微批量处理的情况下，一旦交付，它们将如何对端到端做出贡献。第二部分是关于接收器的部分，而最后一部分则是在一个示例中总结了所有理论要点。 数据源 就正好一次处理而言，源必须是可重播的。也就是说，它必须允许跟踪当前的读取位置，还必须从上一个故障位置开始重新处理。这两个属性都有助于在任意失败（包括 driver 或 executor）后恢复处理状态。可重播源的一个很好的例子是 Apache Kafka 或其基于云的同事 Amazon Kinesis。两者都能够跟踪当前读取的元素-带偏移量的 Kafka 和带序列号的 Kinesis。不可重播源的一个很好的例子是 org.apache.spark.sql.execution.streaming.MemoryStream，在关闭应用程序后由于数据存储在易失性内存中而无法恢复。 借助检查点机制，可以跟踪已处理的偏移量。在结构化流中，检查点文件中存储的项主要是有关当前批次中处理的偏移量的元数据。检查点存储在 checkpointLocation 选项或 spark.sql.streaming.checkpointLocation 配置条目中指定的位置。 对于微批量执行，检查点将集成在以下架构中： 检查点位置通过 StreamingQueryManager的startQuery 和 createQuery 方法从 DataStreamWriter#startQuery() 方法传递到 StreamExecution 抽象类。 StreamExecution 初始化对象 org.apache.spark.sql.execution.streaming.OffsetSeqLog。该对象表示 WAL日志，该日志记录每个已处理批次中存在的偏移量。 该字段表示处理数据源中偏移量的逻辑。当前微批处理的偏移量（我们称其为 N）始终在处理完成之前写入。该事实还假定来自先前微批处理（N-1）的所有数据均已正确写入输出接收器。在以下来自 org.apache.spark.sql.execution.streaming.MicroBatchExecution#constructNextBatch() 的代码段中进行了表示： updateStatusMessage("Writing offsets to log") reportTimeTaken("walCommit") { assert(offsetLog.add( currentBatchId, availableOffsets.toOffsetSeq(sources, offsetSeqMetadata)), s"Concurrent update to the log. Multiple streaming jobs detected for $currentBatchId") logInfo(s"Committed offsets for batch $currentBatchId. " + s"Metadata ${offsetSeqMetadata.toString}") // NOTE: The following code is correct because runStream() processes exactly one // batch at a time.

Apache Kafka Streams 支持流之间的连接，社区对 Apache Spark 期望相同。此功能已在最新的 2.3.0 版本中实现并发布，在此之后的几个月后，现在是讨论该功能的好时机。 该帖子由 2 个部分组成。第一个介绍了流管道中的连接的想法。下一个讨论两种受支持的类型之一-内连接。 连接和流 在流上执行连接操作非常具有挑战性，这有很多原因。最明显的是延迟。对于有界数据源，要连接的所有数据都就位，而对于基于流的数据源，数据则在不断移动。有时（通常？）移动速度不同。可能是由于技术方面的问题（例如摄入管道问题），或者仅是由于功能要求，而相关事件并非总是在相似的时间段内生成。这种功能限制之一可能是在电子商务商店中的订购过程，在该过程中，订单很难很快完成。因此，连接操作必须以某种方式管理相关但非常异步的事件的情况。 解决的另一个重要点是状态管理。由于给定事件的数据可能随时（非常晚）到达，并且存储该事件的存储空间有限，因此引擎必须弄清楚如何处理累积状态，尤其是何时丢弃它。此特定时间对应于我们不希望为给定的连接键接收任何新事件的时刻。 Apache Spark 结构化流提供了连接 2 个或更多流的功能，从而解决了 2.3.0 版本中的两个问题。流到流的连接可以通过以下时间轴来表征： 连接语义与批连接相同 找到匹配元素后立即生成输出（内连接） 水位和时间范围查询用于连接较晚的数据，并确定何时不再发生给定键的其他事件（状态丢弃） 支持不同的连接类型：内连和外连 连接可以级联，即应用于两个以上的流 但是，从 Apache Spark 2.3.1 开始，流到流连接具有几个限制： 它只能与 append 输出模式一起使用（已计划支持其他模式） 连接之前仅支持类似 map 的操作，例如我们不能在连接之前进行聚合 流的内连接 支持的第一种流到流连接是内连接。由于当没有匹配的行不发出时是严格的连接，因此连接的列不需要任何时间限制。但是，这样做很危险，因为即使没有匹配的行也可能在状态存储中保留很长时间。这是因为建议使用一种条件来告知特定键的状态应保留多长时间。下面显示了不清除状态时内连接的最简单情况： it should "output the result as soon as it arrives without watermark" in { val mainEventsStream = new MemoryStream[MainEvent](1, sparkSession.sqlContext) val joinedEventsStream = new MemoryStream[JoinedEvent](2, sparkSession.sqlContext) val stream = mainEventsStream.toDS().join(joinedEventsStream.toDS(), $"mainKey" === $"joinedKey") val query = stream.

以前，我们在 Apache Spark 中发现了流到流的内连接，但是它们不是唯一受支持的类型。另一个是外连接，使我们可以在不匹配行的情况下连接流。 这篇文章是关于结构化流模块中的外连接的。它的第一部分介绍了有关这种连接的一些理论观点。第二篇展示了如何通过一些 Scala 示例来实现它。 当匹配是可选的 流式外连接与经典的，类似批处理的连接没有什么不同。有了它们，我们总是从一侧获得所有行，即使其中一些行在连接数据集中没有匹配项也是如此。对于 RDBMS 之类的有限数据源，此类不匹配将直接返回，其中 null 表示另一侧的行。但是，无限来源的逻辑是不同的。由于不同的特性，例如网络延迟影响或脱机设备产生事件，因此在给定时刻我们可能没有所有连接的元素。因此，我们必须能够将物理连接推迟到我们确定要连接的大多数行都将到来的那一刻。为此，我们需要将一侧的行存储在某处。并且，如果您还记得 Apache Spark 结构化流中流之间的内连接的文章中的一些注释，则 Apache Spark 为此使用状态存储。下图从鸟瞰显示了这一点： 该图像清楚地表明，与内连接水位的情况一样，行被缓冲在状态存储中。 外连接还使用水位和范围查询条件的概念来确定何时不应在给定的行中接收第二个流中的任何新匹配。 这就是为什么完全没有水位的外连接是不可能的： it should "fail without watermark and range condition on watermark in the query" in { val mainEventsStream = new MemoryStream[MainEvent](1, sparkSession.sqlContext) val joinedEventsStream = new MemoryStream[JoinedEvent](2, sparkSession.sqlContext) val mainEventsDataset = mainEventsStream.toDS().select($"mainKey", $"mainEventTime", $"mainEventTimeWatermark") .withWatermark("mainEventTimeWatermark", "2 seconds") val joinedEventsDataset = joinedEventsStream.toDS().select($"joinedKey", $"joinedEventTime", $"joinedEventTimeWatermark") .withWatermark("joinedEventTimeWatermark", "2 seconds") val stream = mainEventsDataset.

上周，我们在 Apache Spark 结构化流中发现了 2 种流对流连接类型。如这些帖子所述，有时可能会省略状态管理逻辑（对于内连接），但通常建议减少内存压力。 Apache Spark 提出了 3 种不同的状态管理策略，这些策略将在以下各节中详细介绍。 这篇文章分为 4 部分。第一个回顾了流对流连接情况下的状态特异性。接下来的 3 个讨论 3 种状态管理策略。 状态和流连接 如 Apache Spark 结构化流中的外连接这篇文章所述，每个潜在的可连接行都缓存在状态存储中。每当找到匹配的行时，都会进行连接并发射结果。内连接类型就是这种情况。对于外连接，逻辑略有不同，由于匹配的行或由于过期状态而发出结果。过期状态表示我们不希望收到给定条目的匹配事件的时刻。希望此行为也适用于内连接，但不同之处在于其可选特性。 没有这种“过期状态”的概念，引擎将使行无限期地匹配，并且由于数据源是无界的，因此不可避免地迟早会失败。因此，Apache Spark 提供了 3 种不同的策略来管理状态过期（水位）。 状态键水位 这些策略中的第一个称为状态键水位。在以下情况下将其应用于查询： 在至少一个连接流中之一中定义了一个水位列-它可以是时间戳列或窗口列。如果仅在一侧定义水位，则 Apache Spark 能够从该水位推导另一侧的水位。 水位列在 JOIN 子句中用作相等约束 此策略的名称来自直接在 JOIN 子句条件中使用水位-因此是状态键。为了说明这一点，我们可以用下面的代码片段举几个例子： "state key watermark" should "be built from watermark used in join" in { val mainEventsStream = new MemoryStream[MainEvent](1, sparkSession.sqlContext) val joinedEventsStream = new MemoryStream[JoinedEvent](2, sparkSession.sqlContext) val mainEventsDataset = mainEventsStream.toDS().select($"mainKey", $"mainEventTime", $"mainEventTimeWatermark") .

在最近关于Apache Spark结构化流的3篇文章中，我们发现了流连接：内部连接，外部连接和状态管理策略。发现所有这些操作背后发生的事情是总结该系列的一个好方法。 这篇文章首先介绍了流连接过程中涉及的类。接下来是专注于与联接有关的状态管理内部的部分。文章以关于连接机制的一小段结尾。 所涉及的类 在流连接所涉及的类中，我们可以区分3个非常重要的类：SymmetricHashJoinStateManager，StreamingSymmetricHashJoinExec 和 StreamingJoinHelper。所有这些都用于流查询执行的不同阶段。 首先，查询的流式表示形式 IncrementalExecution 实例存储对该状态的引用。如果查询具有某些流间连接，则此状态在每次执行时都表示为 StreamingSymmetricHashJoinExec 的实例。该实例在每次执行中都不同，不同点是偏移量统计信息和状态水位谓词。谓词的计算公式为： def getStateWatermarkPredicates( leftAttributes: Seq[Attribute], rightAttributes: Seq[Attribute], leftKeys: Seq[Expression], rightKeys: Seq[Expression], condition: Option[Expression], eventTimeWatermark: Option[Long]): JoinStateWatermarkPredicates 此方法通过应用不同的规则来计算用于从状态存储中丢弃太晚的行的状态水位谓词。 它首先检查查询的相等性 JOIN 中涉及的所有列中是否至少有一个用水位注释标记的列。 如果是，它将自动认为必须将状态键水位策略应用于迟到的行（您可以在 Apache Spark 结构化流的外连接中了解它们）。 如果不是，则检查一个连接侧是否定义了水位列。 如果满足上述条件之一，并且优先选择前者，则使用常规的 org.apache.spark.sql.execution.streaming.WatermarkSupport＃watermarkExpression(optionalWatermarkExpression: Option[Expression], optionalWatermarkMs: Option[Long]) 方法。 JOIN 子句中的相等重要性 流到流连接的实际实现仅接受相等关系作为连接键。 这意味着如果我们有2个流：stream#1(field1[int], field2[timestamp]), stream#2(field10[int], field20[timestamp]), ，则只有 field1 和 field10 之间的相等关系以及 field2 和 field20 将被视为连接键。 如果在 JOIN 的 ON 部分中表示不等式，则会将其转换为 WHERE 条件。 例如，以下查询： val mainEventsDataset = mainEventsStream.toDS().select($"mainKey", $"mainEventTime", $"mainEventTimeWatermark", window($"mainEventTimeWatermark", "3 seconds").

前段时间，我对 Apache Spark 结构化流中 Apache Kafka 连接器的实现提出了3个有趣的问题。 我将在这篇文章中回答他们。 您可以想象，该帖子分为3个部分。 每个人都会回答一个问题。 在文章结尾，您应该更好地了解谁负责 Apache Kafka 连接器中的内容。 问题1：偏移量跟踪 第一个问题是关于 Apache Kafka 偏移量跟踪的。 谁跟踪他们，driver 或 executor？ 在深入探讨该问题之前，让我们回顾一下我在分析结构化流式 Kafka 集成-Kafka源帖子中介绍的一些基础知识。 Apache Kafka 源代码首先从 driver 读取要处理的偏移量，然后将其分配给 executor 以进行实际处理。 因此，我们可以推断出，从这个角度来看，偏移量是由 driver 跟踪的。 您会注意到在 KafkaSource 类内部创建 KafkaSourceRDD 的代码中： // Calculate offset ranges val offsetRanges = topicPartitions.map { tp => // ... } // Create an RDD that reads from Kafka and get the (key, value) pair as byte arrays.

不久前，Sunil 询问我是否可以像基于 DStream 的 API 一样在 Apache Spark 结构化流中加载初始状态。 由于回应并不明显，因此我决定调查并通过这篇文章分享调查结果。 文章首先简短地回忆了 Apache Spark Streaming 模块中的状态初始化。 下一节将讨论可用于在 Apache Spark 结构化流库中执行相同操作的方法。 流中的初始化状态 在基于 DStream 的库中初始化状态很简单。 您只需要创建一个基于键的 RDD 并将其传递给 StateSpec 的 initialState 方法： "streaming processing" should "start with initialized state" in { val conf = new SparkConf().setAppName("DStream initialState test").setMaster("local[*]") val streamingContext = new StreamingContext(conf, Durations.seconds(1)) streamingContext.checkpoint("/tmp/spark-initialstate-test") val dataQueue = new mutable.Queue[RDD[OneVisit]]() // A mapping function that maintains an integer state and return a UserVisit def mappingFunction(key: String, value: Option[OneVisit], state: State[UserVisit]): Option[(String, String)] = { var visitedPages = state.

当我第一次听说 foreachBatch 功能时，我以为这是结构化流模块中 foreachPartition 的实现。但是，经过一些分析，我发现我错了，因为此新功能解决了其他但也很重要的问题。您会发现更多。 在 Apache Spark 2.4.0 功能系列的这一新文章中，我将展示 foreachBatch 方法的实现。在第一部分中，我将简要介绍有关此功能的要点。我还将在其中添加有关实现的一些详细信息。在接下来的2部分中，我将展示.foreachBatch 数据接收器解决的问题。 定义 在 2.4.0 发行版之前，foreach 是我们可以放置一些自定义逻辑的单一接收器。它很容易使用，因为它看起来像包装在类中的 foreach 循环。另外，foreach 接收器非常适合连续执行，因为我们将重点放在每次一行所带来的信息上。但是，由于基于微批处理的管道的适应性通常更差一些，因为我们经常需要对整个累积的微批处理进行某些处理。 2.4.0 版本使用新的 org.apache.spark.sql.execution.streaming.sources.ForeachBatchSink 接收器解决了微批处理的这些问题。它的主要思想很简单。引擎累积在给定的微批次中处理的数据，并将其作为数据集传递到接收器。这不仅意味着您可以对整个数据应用一种逻辑，而且还可以在流传输管道中执行一些纯批处理，例如将数据写入不可流式的数据存储中。 除了具体的数据集之外，foreachBatch 消费者方法还接受一个名为 batchId 的属性。此参数包含生成数据集的微批处理的 ID。您可以使用此属性实现一次传递语义，因为默认情况下，引擎以至少一次语义运行。 最后，由于 ForeachBatchSink 解决了微批次问题，因此您不能将其与连续触发器一起使用。在 DataStreamWriter 内部进行的简单检查显示： } else if (source == "foreachBatch") { assertNotPartitioned("foreachBatch") if (trigger.isInstanceOf[ContinuousTrigger]) { throw new AnalysisException("'foreachBatch' is not supported with continuous trigger") } 您还可以看到接收器不支持分区管道（.partitionBy(...)）。 如果您对引入此逻辑的任务感兴趣，可以在“另请阅读”部分中找到一个链接。 用例：不流式接收器 我已经在上一节中提到了 foreachBatch 的第一个用例。 当您要将处理后的数据保存到关系数据库或键值存储之类的不可流式接收器中时，此新接收器很有用。 为了简单起见，我将使用内存中单例键值存储： "foreachBatch" should "save the data into a key-value memory store" in { val inputStream = new MemoryStream[Int](1, sparkSession.

这些年来，Apache Spark 的流媒体被认为是与微批处理一起工作的。但是，版本 2.3.0 试图对此进行更改，并提出了一种称为“连续”的新执行模型。即使它仍处于实验状态，还是值得更多了解它。 这篇文章介绍了连续流处理的新实验功能。第一部分解释并与微批执行策略进行比较。下一个显示一些内部细节。在第三部分中，我们可以了解至少一次保证。最后一部分在基于比率的源示例中显示了一个简单的连续执行用例。 连续流 毕竟为什么要替换微批​​处理的 Spark 区别标记（与允许连续流处理的 Apache Flink 或 Apache Beam 不同）？答案很简单-延迟。使用微批处理时，延迟很高。在最坏的情况下，等待时间是批处理时间与任务启动时间的总和，估计为几毫秒。连续处理将数据到达与其处理之间的等待时间减少到几毫秒。延迟减少是此（仍）实验特性背后的主要动机。 从鸟瞰来看，连续查询在多个线程中执行。每个线程负责不同的分区。因此，为了保证最佳的并行性，集群中可用核心的数量必须等于要处理的分区的数量。在处理期间，这些线程将结果连续写入结果表。 从技术上讲，可以通过 .trigger(Trigger.Continuous("1 second")) 使用连续触发来启用连续处理。但必须强调的是，在当前（2.3.0）Spark 版本中，并非所有操作都暴露于此模式下。在可用的转换中，我们只能区分投影（select，映射函数）和选择（过滤器子句）。 但是，降低延迟并不是没有代价的。实际上，更快的处理将交付保证从正好一次下降到了至少一次。因此，对于处理延迟比交付保证更为重要的系统，建议执行连续执行。 ContinuousExecution 类 最初，在结构化流中，StreamExecution 包含整个执行逻辑。仅在 2.3.0 版本中，它才成为由 MicroBatchExecution 和 ContinuousExecution 扩展的抽象类。这两个类的出发点都是 org.apache.spark.sql.execution.streaming.StreamExecution#runStream()，在执行一些初始化步骤之后，该类将激活每个执行策略中实现的流查询，方法为 runActivatedStream(sparkSessionForStream: SparkSession)` 方法。 由于以下代码，ContinuousExecution 连续运行查询： do { runContinuous(sparkSessionForStream) } while (state.updateAndGet(stateUpdate) == ACTIVE) 在此 runContinuous 方法内部，发生了许多操作。第一个步骤包括将查询逻辑计划转换为一系列 ContinuousReaders。此接口定义是否可以连续方式读取给定的数据源。当前仅支持 Apache Kafka 和基于速率的源（该源以固定的Y间隔每秒生成 X（行，计数器）行，其中 X 和 Y 是配置参数）。如果某些源或操作不支持连续模式，则在启动查询之前会引发 UnsupportedOperationException。 在创建源之后，将解析起始偏移量，以确定何时开始执行给定执行的处理。如果查询是第一次启动，则偏移量自然为空。但是，再次执行查询时，将从提交日志中检索偏移量。稍后，将为每个映射到读取偏移量的数据源创建一个 StreamingDataSourceV2Relation 实例。接下来，创建数据读取器和写入器，并以与微批处理方法完全相同的方式生成执行计划。 在下一步中，引擎通过 EpochCoordinator 实例以纪元（与微批处理中的批处理类似）进行播放。这是连续处理的杰作。此 RPC 端点负责处理以下消息： 生成新的纪元ID-通过消息 IncrementAndGetEpoch，该纪元 ID 会自动增加。然后，在 runContinuous 方法中返回新值。 返回当前纪元ID-表示为 GetCurrentEpoch 实例的此 getter 消息由定期执行 EpochPollRunnable 中定义的逻辑的线程使用。在此逻辑中，引擎检索当前已知的纪元并将其添加到要处理的消息的内部队列中。在此列表下面的模式中将对此进行详细说明。 在纪元下提交分区-在给定时期内对分区的处理终止时，会将 CommitPartitionEpoch 消息发送给 EpochCoordinator 以发出信号。如果所有纪元的分区都已处理，则 ContinuousExecution 将此纪元标记为已提交。这意味着该纪元被保留在提交日志中，并且在重新处理的情况下，与之关联的偏移量被用作起始偏移量（请参见上一段）。 这些涉及纪元协调器的连续处理逻辑可以在以下简化模式中恢复：

有关 Apache Spark 2.4.0 功能的系列继续进行。在上周发现了桶剪枝之后，现在该切换到结构化流模块并查看其主要进展了。 这篇文章介绍了 2.4.0 版本中添加的水位配置功能。第一部分解释了它的变化。第二个给出了实现细节，其中显示了用于保证向后兼容性的策略之一。最后一部分包含说明水位配置功能的测试用例。 最小或最大水位？ 2.3.0 版可以连接流。如 Apache Spark 结构化流中的流之间的内联中所述，这样做不是一件容易的事，因为联接的流可能具有不同的延迟。水位是控制哪些数据可以联接的方法之一，即使它来得很晚。在 2.4.0 发行版之前，定义了多个水位时，默认情况下使用最小值。它使我们能够以最慢的流移动，并且不会丢失太多数据： 通过 Apache Spark 2.4.0 中引入的更改，我们可以在这种情况下配置使用的水位，并采用最小或最大水位。 当然，选择最大值将导致应用程序以最快的速度向前移动，从而丢失更多数据： 水位配置实现 水位管理是通过 spark.sql.streaming.multipleWatermarkPolicy 配置条目实现的。它采用“最小”或“最大”值之一。它们中的每一个都导致相应的 MultipleWatermarkPolicy 实例的初始化：MinWatermark 和 MaxWatermark。两者都附带一种解决使用的水位值的方法。根据使用的策略，此方法仅是对 Seq 的 min 或 max 方法的调用。负责调用已配置策略的类为 WatermarkTracker。 向后兼容性由 MultipleWatermarkPolicy 条目的默认值 ‘min’ 保证。除非您明确更改此值，否则您应该能够在新的 Spark 版本上运行旧的流传输管道，而不会出现问题。甚至那些从 Apache Spark 2.4.0 之前的检查点恢复的管道。而且由于我们在谈论检查点，所以要记住的一件事很重要。从检查点还原管道后，我们无法修改水位策略。任何更改都将被忽略。 最大水位示例 为了看到这个新功能的实际效果，我们将执行2个测试用例。前者将具有最小水位，因此可以接受更多迟到数据： object FirstWatermark { var FirstKnownValue = "" } def launchDataInjection(mainEventsStream: MemoryStream[MainEvent], joinedEventsStream: MemoryStream[JoinedEvent], query: StreamingQuery): Unit = { new Thread(new Runnable() { override def run(): Unit = { val stateManagementHelper = new StateManagementHelper(mainEventsStream, joinedEventsStream) var key = 0 val processingTimeFrom1970 = 10000L // 10 sec stateManagementHelper.

长查询的重点之一是跟踪。知道查询的执行方式总是很重要的。在结构化流中，由于有了称为 ProgressReporter 的特殊对象，我们可以追踪此执行。 在本文中，我们将重点介绍 ProgressReporter 对象收集的指标。在第一部分中，我们将解释其生命周期以及一些实现细节。下一部分将涵盖公开的信息，而最后一部分将通过一些测试展示 ProgressReporter 的行为。 ProgressReporter 首先，让我们定义这个著名的 ProgressReporter。它是 org.apache.spark.sql.execution.streaming 中的一个 trait，由 StreamExecution 抽象类直接继承，因此由其实现间接继承：org.apache.spark.sql.execution.streaming.continuous.ContinuousExecution 和 org.apache.spark.sql.execution.streaming.MicroBatchExecution。 ProgressReporter 的作用是提供一个接口，该接口一旦实现，便可以自由用于报告有关流查询执行情况的统计信息。 ProgressReporter 定义了严格的生命周期阶段： 1、一切开始于流查询触发器（处理时间或事件时间）执行时。触发器要做的第一件事是对 ProgressReporter 的 startTrigger 方法的调用。此方法使报告程序准备好累积刚刚开始执行的统计信息。 2、稍后，根据选择的流模式（微批或连续），报告器将记录有关几个不同步骤执行情况的统计信息。下一部分将详细介绍这些步骤。为此，它使用了方法 reportTimeTaken[T](triggerDetailKey: String)(body: => T)，该方法将有关执行这些步骤的度量添加到 currentDurationsMs: mutable.HashMap[String，Long]` 字段中。 3、下一步是数据处理，报告者还将收集一些统计信息。 4、将这些统计信息添加到 currentDurationMs 映射后，如果执行微批处理，则 ProgressReporter 调用 finishTrigger(hasNewData: Boolean)。此方法完成触发器的执行，并创建保存执行统计信息的对象，这些统计信息放入 progressBuffer = new mutable.Queue[StreamingQueryProgress]() 中。之后，客户端可以通过公共访问器方法直接从那里检索更新（或最后一个更新）。 在 ProgressReporter 中，我们还可以找到其他一些指标，例如： newData - 它是一个 Map[BaseStreamingSource, LogicalPlan]，其中包含每个源的最新数据。 availableOffsets - 这是一个类似于 map 的 StreamProgress 对象，用于存储未提交到接收器的可用于处理的偏移量。 commitOffsets - 类似于 availableOffset。不同之处在于，它存储已处理和已提交数据的偏移量。 currentBatchId - 当前处理批次的 ID。 currentStatus - org.

https://www.waitingforcode.com/apache-spark-structured-streaming/stateful-aggregations-apache-spark-structured-streaming/read 最近，我们发现了用于处理结构化流中状态聚合的状态存储的概念。但是那时我们还没有花时间在这些聚合上。按照承诺，现在将对其进行描述。 这篇文章开始于聚合与有状态聚合之间的简短比较。在本文结尾之后，直接进行了一些测试，这些测试说明了状态聚合。 聚合与有状态聚合 聚合是从多个值中计算单个值的操作。在 Apache Spark 中，这种操作的示例可以是 count 或 sum 方法。在流式应用程序上下文中，我们谈论有状态聚合，即具有这些值的状态的聚合在一段时间内逐渐增长。 在对结构化流中的输出模式，状态存储和触发器给出了解释之后，就更容易理解状态聚合的特殊性。如所告知的，输出模式不仅确定数据量，而且还决定是否可以丢弃中间状态（如果与水印一起使用的话）。然后，所有这些中间状态都保留在容错状态存储中。上一次触发器执行后，触发器会按固定的时间间隔执行状态计算，这些数据是从上一次触发器执行中累积的。以下架构显示了所有这些部分如何协同工作： 因此，为了简单起见，我们可以将状态聚合定义为结果随时间变化的聚合。 结果计算由触发器启动，并保存在状态存储中，在经过处理的数据通过水印之前（如果已定义），可以从状态存储中删除结果。 有状态聚合示例 两项测试显示了 Apache Spark 结构化流中的状态聚合： "stateful count aggregation" should "succeed after grouping by id" in { val testKey = "stateful-aggregation-count" val inputStream = new MemoryStream[(Long, String)](1, sparkSession.sqlContext) val aggregatedStream = inputStream.toDS().toDF("id", "name") .groupBy("id") .agg(count("*")) val query = aggregatedStream.writeStream.trigger(Trigger.ProcessingTime(1000)).outputMode("update") .foreach( new InMemoryStoreWriter[Row](testKey, (row) => s"${row.getAs[Long]("id")} -> ${row.getAs[Long]("count(1)")}")) .start() new Thread(new Runnable() { override def run(): Unit = { inputStream.

如果你使用 Elasticsearch 6.x 或者更高版本, 那么用 Spark Structured Streaming 写入到 Elasticsearch 就很直截了当。 不像 5.x 之前的版本那样, 为了写入到 Elasticsearch, 你必须实现一个自定义的 sink, 6.x 版本在 Spark Structured Streaming 里面开箱即用。 依赖如下: <dependency> <groupId>org.elasticsearch</groupId> <artifactId>elasticsearch-spark-20_2.11</artifactId> <version>6.2.4</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_${scala.main}</artifactId> <version>2.3.1</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_${scala.main}</artifactId> <version>2.3.1</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming_${scala.main}</artifactId> <version>2.3.1</version> </dependency> 然后写入到 Elasticsearch 就相当简单了。 1、创建带有 Elasticsearch 配置的 SparkSession val spark = SparkSession .builder // run app locally utilizing all cores .master("local[*]") .appName(getClass.getName) .config("es.nodes", "localhost") .config("es.port", "9200") .

Spark Structured Streaming 允许我们以一种直接了当的方式在滑动事件时间窗口上执行聚合。 Spark Structured Streaming 提供开箱即用的事件时间上的窗口操作和水位。 用例 在会话持续时间内求用户收益的总和，当流中有存款事件时，存入当前余额并关闭会话。 窗口和水位 默认情况下，窗口和水位与基于时间的窗口一起使用。在我们的用例中，这不起作用，因为当我们收到一个存放当前余额的信号时，我们想关闭会话。 结果 我们可以通过基于 mapGroupsWithState 函数实现自定义会话窗口来实现. 定义自定义会话类 case class Transaction(sessionId: String, winAmount: Double, deposit: Boolean) case class SessionTrackingValue(totalSum: Double) case class SessionUpdate(sessionId: String, currentBalance: Double, depositCurrentBalance: Boolean) 创建本地 SparkSession val spark: SparkSession = SparkSession .builder .master("local[*]") .appName(getClass.getName) .getOrCreate() 创建 socket 流 val socketStream: DataFrame = spark.readStream // socket as stream input .format("socket") // connect to socket port localhost:9999 waiting for incoming stream .

https://www.waitingforcode.com/apache-spark-sql/apache-spark-2.4.0-features-bucket-pruning/read 桶定义 均衡的分区让我们更快地处理数据。例如，我们可以收集物联网事件，并把它们按照日期分区并存储在树一样的结构中： /events/2018/10/29 /events/2018/10/30 /events/2018/10/31 如果我们想按物联网设备号划分相同的数据，我们该怎么办？ 从技术上讲，这是可行的，但从概念上讲，它可能不如基于日期的分区有效。 设备密钥是具有非常高的基数（可能的唯一值的数量）的值，我们最终将得到一棵具有数百或数千个子目录的树。 对于不是分区的最佳选择的值问题，解决方案之一是桶存储，也称为群集。 桶是“分区内的分区”。 区别在于桶的数量是固定的。 在大多数情况下，这些值会使用基于哈希的简单策略分配给存储桶。 您可以在下面找到 Apache Spark 中存储桶的示例： "Spark" should "create buckets in partitions for orders Dataset" in { val tableName = s"orders${System.currentTimeMillis()}" val orders = Seq((1L, "user1"), (2L, "user2"), (3L, "user3"), (4L, "user1")).toDF("order_id", "user_id") orders.write.mode(SaveMode.Overwrite).bucketBy(2, "user_id").saveAsTable(tableName) val metadata = TestedSparkSession.sessionState.catalog.getTableMetadata(TableIdentifier(tableName)) metadata.bucketSpec shouldBe defined metadata.bucketSpec.get.numBuckets shouldEqual 2 metadata.bucketSpec.get.bucketColumnNames(0) shouldEqual "user_id" } 在分布式数据处理框架中，这种方法往往有助于避免洗牌阶段。例如，当桶用于2个数据集与 Spark SQL 中排序合并连接时，因为这两个数据集已经可以位于同一个分区中, 因此洗牌可能没有必要。当然，这两个数据集必须具有相同的分区数并使用哈希分区算法。 桶剪枝实现 在 Apache Spark 2.4.0 之前，当桶列之一参与查询时，Spark 引擎并没有作出任何优化。毕竟，由于桶存储是确定性的，引擎只能读取桶文件存储筛选值。

mapGroupsWithState 解释 mapGroupsWithState(timeoutConf: GroupStateTimeout)(func: (K, Iterator[V], GroupState[S]) => U) 是一个应用到数据组上的转换。由于它要求数据分组, 所以可能会引入 shuffling 的风险。通过调用 func 参数来计算 state。这个转换可以用在有界数据源和无界数据源上。对于有界数据源, 最终状态会立即被计算出来, 对于无界数据源, 每次触发处理后, 状态都可能会改变。 mapGroupsWithState 的参数 timeoutConf: GroupStateTimeout 这个参数负责处理超时。 mapGroupsWithState 转换接收 3 个不同的超时值: NoTimeout, ProcessingTimeTimeout 和 EventTimeTimeout(只在设置了水位时才有效)。状态过期时间定义在作为 func 参数值的函数里面, 它要么是 org.apache.spark.sql.streaming.GroupState#setTimeoutDuration (processing time), 要么是 org.apache.spark.sql.streaming.GroupState#setTimeoutTimestamp (event time)。 每次至少有一条记录被发送到组里时, 就更新超时时间。状态只有在给定组在指定阈值时间内未接收到任何数据时才被认为过期。 超时时间是在指定 GroupState 对象里面定义的, 所以根据处理的组指定不同的超时配置也是完全可行的。 func: (K, Iterator[V], GroupState[S]) => U - 该函数定义如何处理组里的值以生成状态。 如你所见, 它完全是类型无关的, 例如, 值([V]) 和状态 [S] 都不影响所返回的数据集([U])的类型。 当两个条件之一满足时, 该函数就会被调用: 要么组里有要处理的新值, 要么状态已经过期。在状态过期这种情况下, 会使用一个空列表的值作为第二个参数来调用该函数。 GroupState 另一个重要的与状态管理有关的对象是 org.

标准 Raku 中, for 关键字是没有 else 分支的, 但是 Raku 可以实现 for...else 方言, 新建一个模块, 命名为 Slang::ForElse, 其目录结构如下: Slang/ForElse.pm6 其中 ForElse.pm6 的内容如下: role ForElse::Grammar { rule statement_control:sym<for> { <sym><.kok> {} <.vetPerl5Syntax> <xblock(2)> {} [ 'else' <elseblock=pblock> ]? } rule vetPerl5Syntax { [ <?before 'my'? '$'\w+\s+'(' > <.typed_panic: 'X::Syntax::P5'> ]? [ <?before '(' <.EXPR>? ';' <.EXPR>? ';' <.EXPR>? ')' > <.obs('C-style "for (;;)" loop', '"loop (;;)"')> ]? } } role ForElse::Actions { use nqp; use QAST:from<NQP>; sub lookup(Mu \match, \key) { nqp::atkey( nqp::findmethod(match, 'hash')(match), key ).

第18周挑战赛的第一个任务是在一组字符串中找出最长的共同子串，即给定一组字符串如 “ABABC”、“BABCA “和 “ABCBA”，打印出 “ABC”。 “最佳实践"解决方案是一种被称为后缀树的技术，它需要一些中等复杂的编码。然而，我们可以使用一种简单得多的方法，对长达几十万字符的字符串获得非常合理的性能。 让我们从一个非常简洁，但次优的解决方案开始。如果我们有一个每个字符串的所有可能的子串的列表: ABABC: ABABC, ABAB, BABC, ABA, BAB, ABC, AB, BA, BC, A, B, C BABCA: BABCA, BABC, ABCA, BAB, ABC, BCA, BA, AB, BC, CA, B, A, C ABCBA: ABCBA, ABCB, BCBA, ABC, BCB, CBA, AB, BC, CB, BA, A, B, C …那么我们可以把每个列表当作一个集合，取这些集合的交集: ABC, AB, BA, BC, A, B, C …然后只需找到最长的元素: ABC 当然，有时所有可能的子串的交集会有一个以上的最长元素: SHAMELESSLY NAMELESS LAMENESS …所以我们需要确保我们的算法能够找到所有的子串。 鉴于 Raku 语言中内置了集合和集合操作，这里唯一的挑战就是为每个字符串生成所有可能的子串的完整集合。而且，鉴于 Raku 有非常复杂的正则表达式匹配，这也不是什么真正的挑战。 在普通的 regex 匹配中:

狂热的，贪婪的本性是太少 20 年前，我首次涉足 Perl 编程的工作之一是使用一种纯文本工具，分析其结构并为给定的线宽整齐地设置其格式的工具。 这是一个中等复杂的换行应用程序，我每天都使用它来整理电子邮件通信，软件文档和博客条目。 因此，第 19 届 Perl 每周挑战赛的第二项任务-实现"贪婪的"换行算法-在许多方面对我来说是一个老朋友。 贪婪的换行只是将输入文本中的每个单词都添加到输出的当前行中，除非这样做会导致输出行超出所需的最大行宽，在这种情况下，它会在该点断开行并继续填充第二行，等等。 因此，一个 45 列的贪婪包装的段落如下所示： It is a truth universally acknowledged, that a single man in possession of a good fortune must be in want of a wife. However little known the feelings or views of such a man may be on his first entering a neighbourhood, this truth is so well fixed in the minds of the surrounding families, that he is considered the rightful property of some one or other of their daughters.

http://blogs.perl.org/users/damian_conway/2019/08/with-friends-like-these.html C-o-rr-a-ll-i-n-g d-i-tt-o-e-d l-e-tt-e-r-s 我本打算本周重点关注第20周挑战的第一个任务……但我能说什么呢？这个任务是把一个在命令行中指定的字符串分解成相同的字符: "toolless" → t oo ll e ss "subbookkeeper" → s u bb oo kk ee p e r "committee" → c o mm i tt ee 但是这在 Raku 中就是小菜一碟: use v6.d; sub MAIN (\str) { .say for str.comb: /(.) $0*/ } 或者使用更优雅的方式： .say for $str.comb: /\w+ % <same>/ 而在 Perl 中也几乎一样简单: use v5.30; my $str = $ARGV[0] // die "Usage:\n $0 <str>\n"; say $& while $str =~ /(.

在我过去写的一些博客中, 我经常遇到 Raku 不能如我所愿的情况。这是件小事儿, 但是像许多其它的小事儿一样, 它不经常发生但确一直刺激到我。 在我之前写的东西中, 我使用了一些 Raku 代码完美地阐述了这一点。我需要为 @values 数组中的每个值创建一个对象, 如果 @values 为空, 则创建一个特殊的对象: for @values Z $label,"",* -> ($value, $label) { Result.new: desc => "$label ($param.name() = $value)", value => timed { $block($value) }, check => { last if .timing > TIMEOUT } } if !@values { Result.new: desc => $label, value => timed { $block(Empty) } } 几乎在同一时间, 在其它我所写的代码中(不在博客里的), 我需要完全相同的结构…处理数组中的每个元素, 如果数组中没有元素, 则做一些不同的处理: for @errors -> $error { note $error if DEBUG; LAST die X::CompilationFailed.

第 15 周挑战赛的第一个任务是找到甜品的最佳排队位置。 这是对创造越野车鞭子或牛仔电影或实体书店或碳基能源的老难题的重新表述：什么时候才是选择在一个不断缩小的馅饼中获得一块不断增加的最佳时机？ 在这个任务中，馅饼足够大，可以容纳100个人，但排在第一位的人只能得到相当于其中1%的一块。第二个人得到剩下的2%，第三个人得到之后剩下的3%，等等，等等，直到第一百个人得到前面99个人分到自己那份后剩下的100%的碎屑。换句话说，我们要"让苹果派再次伟大！" 实际的任务是确定哪一个是队伍中的最佳位置。是早早地进入，得到一小块还算大的馅饼呢？还是等待，最终获得一块更大的剩饼？让我们用六种不同的方式来回答这个问题。 必需的切片 我的开发者生涯是以C语言程序员的身份开始的，我仍然觉得以命令式模式来处理像这样的小问题是很自然的。所以，我第一次尝试回答这个问题（幸好是用 Raku 而不是C语言）的时候是这样的。 # Start with a full pie, and no idea who gets the most... my $pie = 1.00; # i.e. 100% my $max-share = 0.00; # i.e. 0% my $max-index; # For everyone in line... for 1..100 -> $N { # Take N% of what’s left as the next share... my $share = $pie * $N/100; $pie -= $share; # Remember it if it’s maximal.

有一个解析报文的 Java 程序, 里面封装好了可用的函数, 用来解析车辆上的原始报文, kafka 数据如下所示(rawData 字段为报文)。我们用它提供的 libs 来解析该报文。 {"keyId":"XXXXXXXXXXXXX","sampleTime":"1563348447814","msgType":"gbdata","dataList":{"msgType":"gbdata","vehicleModelCode":"1xd9","vehicleSeries":"A76","vin":"LLLLLLLLLLLLLL","rawData":"131302fe4c4e414132414131354b355331303236330101271307110f1b190102010100000000546d0eae26c063010f03e80000050006c3a1fd015f5e080601431018010c100801184a010344070000000000000000000801010eae26c0005c00015c100f101310111010101110101014100b10121010100f1008100e100c100c1010101210111013100e100c1010101310101015101310131015101410141013100e100c100e100d1010100f10131013101410131015100c100d100d100f100e100e101210161015101510431016101710131016101510151016100f101310131016101210101018101010151014101710141014101310131017100e10101013100e100f10111013101410141017101410181016101510161017090101002048474445454545454546444545454545454645464545494a4949464849484849bb","sampleTime":"1563348447814"}} 要在新的项目中使用它们的库, 需要做的是: 在 src 同级目录下, 创建一个 libs 目录, 把 xx.jar 复制到 libs 目录里面 在 src 同级目录下, 创建一个 out 目录, 用于 build Artifact 时存放最后的 jar 包文件 File -> Project Structure -> Artifacts -> + -> Extracted Directory -> 选择外部库.jar File -> Project Structure -> Artifacts -> 勾选 include in projects. (重要) File -> Project Structure -> Libraries -> + -> 选择 libs 目录和外部库.

练习 1 本练习将让您在 NQP 中执行一些简单的任务，以熟悉基本语法。 Hello, world 最简单的事情: say('Hello, world'); 现在把它弄错: say 'Hello, world'; 请注意你将如何大声地抱怨。 变量 将值绑定到标量。 使用 say 来输出它. 尝试使用赋值代替绑定。 观察它的失败。 声明一个数组。 使用索引, 例如 @a[0], 去绑定并访问元素。 注意在 Raku 中, 符号是不变的; 使用 $a[0] 则会抱怨未声明的 $a! 尝试绑定一个数组字面量, 例如 [1,2,3] 到一个 @arr 变量上。现在把该数组字面量绑定到一个 $scalar 变量上。注意观察索引是怎样在这两种结构上都工作良好的。 声明一个哈希变量, %h. 尝试使用字面值语法索引 (%h<key>) 和花括号语法索引 (%h{'key'})。 循环 使用 for 来迭代数组, 一次迭代一个元素, 先使用 $_ 迭代, 然后使用 pointy block (for @a -> $val { ... }) 迭代。 你可以一次迭代两个元素吗？ 试试看！ 迭代散列变量, 打印出该散列每个 pair 对儿的 .

上午介绍了 Spark 的用户自定义聚合函数, 其实是官网的例子, 它使用 Spark SQL。下午我们来看一个稍微复杂点的例子, 这次我们在 agg 中使用自定义的聚合函数。 我们的程序读取 JSON数据, 其中 vin 是车架号; timestamp 是信号发生时间, 信号每隔 1 秒发生一次; PtcConsmp 是一个 Long 类型的值。 {"vin":"KLSAP5S09K1214720","timestamp":1543766400000,"PtcConsmp":5000} {"vin":"KLSAP5S09K1214720","timestamp":1543766401000,"PtcConsmp":5000} {"vin":"KLSAP5S09K1214720","timestamp":1543766402000,"PtcConsmp":5000} {"vin":"KLSAP5S09K1214720","timestamp":1543766403000,"PtcConsmp":5000} {"vin":"KLSAP5S09K1214720","timestamp":1543766405000,"PtcConsmp":5000} {"vin":"KLSAP5S09K1214720","timestamp":1543766406000,"PtcConsmp":850} {"vin":"KLSAP5S09K1214720","timestamp":1543766407000,"PtcConsmp":850} {"vin":"KLSAP5S09K1214720","timestamp":1543766408000,"PtcConsmp":850} {"vin":"KLSAP5S09K1214720","timestamp":1543766409000,"PtcConsmp":850} {"vin":"KLSAP5S09K1214720","timestamp":1543766410000,"PtcConsmp":850} {"vin":"KLSAP5S09K1214720","timestamp":1543766411000,"PtcConsmp":850} {"vin":"KLSAP5S09K1214720","timestamp":1543766412000,"PtcConsmp":850} {"vin":"KLSAP5S09K1214720","timestamp":1543766424000,"PtcConsmp":1650} {"vin":"KLSAP5S09K1214720","timestamp":1543766425000,"PtcConsmp":1650} {"vin":"KLSAP5S09K1214720","timestamp":1543766426000,"PtcConsmp":1650} ... 我们规定： 如果一段连续的时间内, PtcConsmp 的值发生变化, 则记为一次 PTC 切换。 如果发生一次时间缺失, 也记为一次 PTC 切换。 计算每个 vin 的 PtcConsmp 切换次数。 我们借助 Spark 的 UDAF 函数来计算这个需求。首先我们定义一个 PTCPwrAgg 对象, 它扩展自 UserDefinedAggregateFunction。所以 PTCPwrAgg 里必须实现如下几个函数: inputSchema 输入数据 bufferSchema 中间状态 dataType 返回值类型 deterministic true initialize 初始化缓冲区 update 更新缓冲区 merge 合并缓冲区 evaluate 计算最终结果 package ohmysummer import org.

聚合 内置的 DataFrame 函数提供常见的聚合，如 count()，countDistinct()，avg()，max()，min() 等。虽然这些函数是为 DataFrame 设计的，但 Spark SQL 也有类型安全的版本其中一些在 Scala 和 Java 中使用强类型数据集。此外，用户不限于预定义的聚合函数，并且可以创建自己的聚合函数。 无类型用户自定义函数 用户必须扩展 UserDefinedAggregateFunction 抽象类以实现自定义无类型聚合函数。例如，用户定义的平均值可能看起来像这样: import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.sql.expressions.MutableAggregationBuffer import org.apache.spark.sql.expressions.UserDefinedAggregateFunction import org.apache.spark.sql.types._ object MyAverage extends UserDefinedAggregateFunction { // 该聚合函数输入参数的数据类型 def inputSchema: StructType = StructType(StructField("inputColumn", LongType) :: Nil) // 聚合缓冲区中值的数据类型 def bufferSchema: StructType = { StructType(StructField("sum", LongType) :: StructField("count", LongType) :: Nil) } // 返回值的数据类型 def dataType: DataType = DoubleType // 对于相同输入该函数是否始终返回相同输出 def deterministic: Boolean = true // 初始化给定的聚合缓冲区。缓冲区本身是一个 "Row"，除了诸如在索引处检索值(例如，get()，getBoolean()) 标准方法之外，提供了更新其值的机会。 // 请注意，缓冲区内的数组和 map 映射仍然是不可变的 def initialize(buffer: MutableAggregationBuffer): Unit = { buffer(0) = 0L // sum 值 buffer(1) = 0L // 元素个数 } // 使用来自 `input` 的新输入数据更新给定的聚合缓冲区 `buffer` def update(buffer: MutableAggregationBuffer, input: Row): Unit = { if (!

动机 将所有内容连接在一起是应用程序开发的一个单调乏味的部分。有几种方法可以将数据、服务和表示类相互连接起来。为了对比这些方法，我们将为披萨订购网站编写支付代码： public interface BillingService { /** * Attempts to charge the order to the credit card. Both successful and * failed transactions will be recorded. * * @return a receipt of the transaction. If the charge was successful, the * receipt will be successful. Otherwise, the receipt will contain a * decline note describing why the charge failed. */ Receipt chargeOrder(PizzaOrder order, CreditCard creditCard); } 伴随着实现，我们会给我们的代码写单元测试。在测试中，我们需要一个 FakeCreditCardProcessor 来避免支付到真实的信用卡。 直接构造函数调用 以下是我们刚刚刷新信用卡处理器和事务记录器时代码的样子：

之前介绍过在安卓手机上运行 Raku, 那是一种很费劲的折腾, 你得下载一个完整的 Ubuntu 系统, 然后自行编译安装 Raku。虽然现在 Docker 很方便, 但是我不想用它。 有了 Termux 就发现下面这个操作更清凉, 更环保。 我们到 https://github.com/its-pointless/gcc_termux 看到里面的 readme 说得语焉不详, 对于刚接触 Termux 的人来说有点毫无头绪, 但是作者依然提供了一个傻瓜式一键设置脚本, 我拿来改造下: #!/data/data/com.termux/files/usr/bin/sh # Get some needed tools. coreutils for mkdir command, gnugp for the signing key, and apt-transport-https to actually connect to the repo # and clang and make for installing Zef apt-get update apt-get --assume-yes upgrade apt-get --assume-yes install coreutils gnupg wget clang make git # Make the sources.

把旧项目改一改, 提交到 git 仓库中, 这是很多程序员的学习手段。但是, 旧项目的 history commit 中往往有很多敏感信息, 例如数据库登录信息。我就比较奇特了, 我误 commit 了一个大文件(jar 包, 囧)。 历史提交信息正好我也不想保留了, 我想把改好的这个版本当作一次完整的原子提交。即当作一个新仓库了。做法如下： 1）切换到一个新分支 git checkout --orphan latest_branch 缓存所有文件(.gitignore中的除外) git add -A 提交跟踪后的文件 git commit -am "init commit" 删除 master 分支 git branch -D master 重命名当前分支为 master git branch -m master 提交到远程 master 分支 git push -f origin master

变量绑定 在 Rust 中, 我们将一些值绑定到一个名字上, 称之为变量绑定。使用 let 来声明一个绑定： fn main() { let x = 5; } 所有权 Rust 中的变量绑定有一个属性: 变量拥有它们所绑定的值的所有权。当绑定超出作用域, 变量绑定的资源就会被释放。例如： fn foo() { let v = vec![1,2,3,4,5]; } 当 v 进入作用域, 会在栈上创建一个新的 vector, 并在堆上为该 vector 的 5 个元素分配空间。当 v 在 foo() 的末尾离开作用域时, Rust 会释放该 vector 占用的资源。 移动语义 在 Rust 中, 一个值一次只能有一个所有者。 let v = vec![1,2,3,4,5]; // ① 声明一个绑定 let s = v; // ② v moved to s println!("v[0] is: {}", v[0]); // ③ error: use of moved value: `v` 除了 ② 里面发生所有权转移(move)外, 另外一种常见的的 move 语义发生在将变量作为参数传递给函数时:

现在这个项目唯一的乐趣就是使用我学到的新技术重构之前的项目。今天完成了数据适配, 该项目目录结构如下: ├── EnApplication.scala ├── NaApplication.scala ├── conf │ ├── HbaseConfiguration.scala │ ├── KafkaConfiguration.scala │ ├── MysqlConfiguration.scala │ └── SparkConfiguration.scala ├── core │ ├── Adapter.scala │ └── impl │ ├── AdapterImpl.scala │ └── EnAdapterImpl.scala ├── model │ ├── EnSourceData.scala │ └── SourceData.scala ├── module │ ├── EnMainModule.scala │ └── MainModule.scala ├── pipeline │ ├── NaSchema.scala │ ├── NaSparkSession.scala │ └── SourceDataFrame.scala └── util └── XsUdfs.scala 与之前的项目一样, 依旧使用了依赖注入: Kafka 配置 class KafkaConfiguration extends Serializable{ private val config: Config = ConfigFactory.

SimpleStringSchema SimpleStringSchema 把 message 反序列化为字符串。如果你的 message 有键, 则忽略键。 val myConsumer = new FlinkKafkaConsumer( kafkaConf.topic, new SimpleStringSchema(), properties ) JSONDeserializationSchema JSONDeserializationSchema 使用 jackson 将 message 反序列化为 json 格式的消息并返回 com.fasterxml.jackson.databind.node.ObjectNode 对象流。你可以使用 .get("property") 方法访问字段。再一次, 键被忽略。 val myConsumer = new FlinkKafkaConsumer( kafkaConf.topic, new JSONDeserializationSchema(), properties ) JSONKeyValueDeserializationSchema JSONKeyValueDeserializationSchema 与前一个非常类似，但处理带有json编码的键和值的消息。 val myConsumer = new FlinkKafkaConsumer( kafkaConf.topic, new JSONKeyValueDeserializationSchema(true), properties ) 返回的 ObjectNode 包含如下字段： key：键中存在的所有字段 value：所有的 message 字段 metadata（可选）：暴露消息的 offset, partition 和 topic （将 true 传递给构造函数以获取元数据） 例如：

有一次, 在威马项目, 项目经理说, xxx, 你能不能给我查一下 kafka 中, 这个车某一时间段的数据？ 我说不好查, 查询 Kafka 中某一条记录, 需要 Consumer 程序来消费。虽然 kt 之类的 Kafka 命令行工具可以查询 Kafka 中数据. 但是能力有限, 它只能从某个 offset 开始查询, 满足不了我们的过滤条件。 幸运的是, ksql 可以让我们像查询 sql 一样来查询 kafka, 毕竟写 sql, 谁不会呢？于是我们到它的官网, 看到 ksql 这个产品的宣传是 Streaming SQL for Apache Kafka。 毕竟底层用的是 Kafka-Streams, 所以 ksql 支持流式查询。我们下载最新的 ksql。我们将程序解压到 ~/opt/confluent-5.0.0 启动 zk cd ~/opt/confluent-5.0.0 bin/zookeeper-server-start -daemon etc/kafka/zookeeper.properties 启动 kafka cd ~/opt/confluent-5.0.0 bin/kafka-server-start -daemon etc/kafka/server.properties 创建 topic 和 data confluent 自带了一个 ksql-datagen 工具，可以创建和产生相关的 topic 和数据，

KSQL Stream Processing CookBook 数据过滤 KSQL 流式查询持续运行。您可以使用 KSQL CREATE STREAM AS 语法将流式查询输出持久保存到 Kafka topic。 KSQL 从一个 Kafka topic 中实时获取事件，转换它们并将它们连续写入另一个 topic。 此示例显示如何从入站 topic 过滤数据流数据以排除源自特定地理位置的记录。 指示 在此示例中，使用名为 purchase 的源事件流。 { "order_id": 1, "customer_name": "Maryanna Andryszczak", "date_of_birth": "1922-06-06T02:21:59Z", "product": "Nut - Walnut, Pieces", "order_total_usd": "1.65", "town": "Portland", "country": "United States" } 1、在 KSQL 中, 注册 purchases 流: ksql> CREATE STREAM purchases \ (order_id INT, customer_name VARCHAR, date_of_birth VARCHAR, \ product VARCHAR, order_total_usd VARCHAR, town VARCHAR, country VARCHAR) \ WITH (KAFKA_TOPIC='purchases', VALUE_FORMAT='JSON'); 2、检查收到的前几条消息：

我们需要把 Flink 安装路径中的 opt 和 lib 添加到项目的 Libraries 中。 依次点击 File -> Project Structure -> Project Settings -> Libraries, 然后点击右侧栏中的加号 ➕, 选择 Java, 在弹出的对话框中选择 Flink 安装路径。 我的 Flink 安装在 /usr/local/Cellar/apache-flink/1.8.0/libexec, 默认是隐藏的, 所以要点击 Macintosh HD, 同时按住 Command + Shift + . 即可显示出隐藏目录, 分别添加 opt 目录和 lib 目录到相应的模块的 Libraries 中。这样就可以在 IDEA 中直接运行 Flink 程序了。 tree /usr/local/Cellar/apache-flink/1.8.0/libexec ├── bin │ ├── config.sh │ ├── start-cluster.sh │ ├── ... ├── conf │ ├── flink-conf.yaml │ ├── log4j-cli.

使用 Flink 重写一下之前用 Spark Streaming 做的项目, 发现 Flink 在实时处理上比 Spark Streaming 友好。 结构 目录结构跟之前基本一样: src ├── main │ ├── java │ ├── resources │ │ └── application.conf │ └── scala │ ├── dataset │ └── datastream │ ├── com │ │ └── gac │ │ └── xs6 │ │ ├── BigdataApplication.scala │ │ ├── conf │ │ │ └── KafkaConfiguration.scala │ │ ├── core │ │ │ ├── Adapter.scala │ │ │ └── impl │ │ │ ├── AdapterImpl.

准备 zef install App::Mi6 # 安装模块骨架工具 mi6 new FakeStreaming # 创建一个模块骨架 使用 tree 看一下目录结构： ├── Changes ├── LICENSE ├── META6.json ├── README.md ├── bin │ └── fake-stream ├── dist.ini ├── lib │ └── FakeStreaming.pm6 └── t └── 01-basic.t 我们主要看 META6.json, bin 目录, lib 目录。 META6.json { "authors" : [ "" ], "build-depends" : [ ], "depends" : ["PKafka"], "description" : "a commandline tool to simulate socket streraming mock and kafka producer", "license" : "Artistic-2.

准备 ① 启动 Flink 本地模式 /usr/local/Cellar/apache-flink/1.7.0/libexec/bin/start-cluster.sh Starting cluster. Starting standalonesession daemon on host summer. Starting taskexecutor daemon on host summer. ② 模拟实时流数据 raku fake-streaming.pl6 该脚本每隔秒打印一行数据, 每隔 15 秒暂停打印, 然后在继续每隔一秒打印一行, 然后在 sleep 15 秒, 模拟行程间隔。 sub MAIN(Str :$host = '0.0.0.0', Int :$port = 3333) { my $vin = 'LSJA0000000000091'; my $last_meter = 0; react { whenever IO::Socket::Async.listen($host, $port) -> $conn { react { my Bool:D $ignore = True; whenever Supply.interval(15).rotor(1, 1 => 1) { $ignore = !

假设有一段文本, =begin code 和 =end code 把文本分割为一个一个的 section, 我想提取每一个 section 之间的内容。 Grammar 来拯救！ my $excerpt = q:to/END/; Here's some unimportant text. =begin code This code block is what we're after. We'll use 'ff' to get it. =end code More unimportant text. =begin code I want this line. and this line as well. HaHa =end code More unimport text. =begin code Let's to go home. =end code END Grammar #use Grammar::Tracer; #use Grammar::Debugger; grammar ExtractSection { rule TOP { ^ <section>+ %% <.

toggle 字面意思是开关, 函数签名如下： method toggle(Any:D: *@conditions where .all ~~ Callable:D, Bool :$off --> Seq:D) 它接收一个数组, 数组中的每个元素都是一个 Callable。 它迭代调用者, 产生一个 Seq, 根据开关是开还是关把接收到的值填充到结果中, 开关是开还是关取决于 @conditions 数组里面调用 Callables 的结果: say (1..15).toggle(* < 5, * > 10, * < 15); # OUTPUT: «(1 2 3 4 11 12 13 14)␤» say (1..15).toggle(:off, * > 2, * < 5, * > 10, * < 15); # OUTPUT: «(3 4 11 12 13 14)␤» 想象一下开关是打开还是关闭（True或False），如果它打开则产生值。默认情况下，该开关的初始状态处于 “on” 位置，除非 :$off 设置为 true 值，在这种情况下，初始状态将为"off"。

grammars Grammar 结合 actions 可以解析字符串并从字符串中产生一些东西。如果说任何压缩后的数据遵循可能由相应的解压缩算法“解析”的结构，那就不足为奇了。 那么为什么不使用 Raku Grammar 进行这类工作呢？特别是我熟悉的压缩。 深入研究任天堂压缩 任天堂在他们的 SNES 游戏中使用了相同的基本压缩格式，并根据游戏使用了一些变体。这非常简单，在 ~2Mhz 的 SNES CPU 上很容易。 它是这样的： <a byte that contains an header> the header is split into 2 parts : the 3 left most bits form a command number The 5 other bits code the lenght associated with the command <An arbitrary number of bytes associated with the command> <repeat until the header is \xFF> 命令取决于游戏，但是 Zelda 3 的命令非常简单

假设有一个 0..100 的数字区间, 需要分成下面这样的区间： 0..5 5..10 10..15 15..20 20..25 25..30 30..35 35..40 40..45 45..50 50..55 55..60 60..65 65..70 70..75 75..80 80..85 85..90 90..95 95..100 rotor 来拯救： .minmax.say for (0..100).rotor(6 => -1) 我想偷懒用程序生成下面的代码： if (current >=0 && current < 5) "0" if (current >=5 && current < 10) "1" if (current >=10 && current < 15) "2" if (current >=15 && current < 20) "3" if (current >=20 && current < 25) "4" if (current >=25 && current < 30) "5" if (current >=30 && current < 35) "6" if (current >=35 && current < 40) "7" if (current >=40 && current < 45) "8" if (current >=45 && current < 50) "9" if (current >=50 && current < 55) "10" if (current >=55 && current < 60) "11" if (current >=60 && current < 65) "12" if (current >=65 && current < 70) "13" if (current >=70 && current < 75) "14" if (current >=75 && current < 80) "15" if (current >=80 && current < 85) "16" if (current >=85 && current < 90) "17" if (current >=90 && current < 95) "18" if (current >=95 && current < 100) "19" for (0.

使用 Structured Spark Streaming 进行行程划分 我们有一辆车, 车上的传感器每隔 1 秒发出一条 JSON 格式的数据。每条数据包含 3 个字段： field description type unit vin 车辆唯一标识 String 辆 createTime 信号发生时间 Long 毫秒 mileage 当前里程数 Int 千米 下面一段样本数据: {'vin':'LSJA0000000000091','createTime':1546667565,'mileage':0} {'vin':'LSJA0000000000091','createTime':1546667566,'mileage':1} {'vin':'LSJA0000000000091','createTime':1546667567,'mileage':2} {'vin':'LSJA0000000000091','createTime':1546667568,'mileage':3} {'vin':'LSJA0000000000091','createTime':1546667569,'mileage':4} {'vin':'LSJA0000000000091','createTime':1546667570,'mileage':5} {'vin':'LSJA0000000000091','createTime':1546667571,'mileage':6} {'vin':'LSJA0000000000091','createTime':1546667572,'mileage':7} {'vin':'LSJA0000000000091','createTime':1546667573,'mileage':8} {'vin':'LSJA0000000000091','createTime':1546667574,'mileage':9} {'vin':'LSJA0000000000091','createTime':1546667575,'mileage':10} {'vin':'LSJA0000000000091','createTime':1546667576,'mileage':11} {'vin':'LSJA0000000000091','createTime':1546667577,'mileage':12} {'vin':'LSJA0000000000091','createTime':1546667578,'mileage':13} {'vin':'LSJA0000000000091','createTime':1546667579,'mileage':14} {'vin':'LSJA0000000000091','createTime':1546667580,'mileage':15} {'vin':'LSJA0000000000091','createTime':1546667581,'mileage':16} {'vin':'LSJA0000000000091','createTime':1546667582,'mileage':17} {'vin':'LSJA0000000000091','createTime':1546667583,'mileage':18} {'vin':'LSJA0000000000091','createTime':1546667584,'mileage':19} {'vin':'LSJA0000000000091','createTime':1546667585,'mileage':20} {'vin':'LSJA0000000000091','createTime':1546667586,'mileage':21} {'vin':'LSJA0000000000091','createTime':1546667587,'mileage':22} {'vin':'LSJA0000000000091','createTime':1546667588,'mileage':23} {'vin':'LSJA0000000000091','createTime':1546667589,'mileage':24} {'vin':'LSJA0000000000091','createTime':1546667590,'mileage':25} {'vin':'LSJA0000000000091','createTime':1546667591,'mileage':26} {'vin':'LSJA0000000000091','createTime':1546667592,'mileage':27} {'vin':'LSJA0000000000091','createTime':1546667593,'mileage':28} {'vin':'LSJA0000000000091','createTime':1546667594,'mileage':29} {'vin':'LSJA0000000000091','createTime':1546667595,'mileage':30} {'vin':'LSJA0000000000091','createTime':1546667596,'mileage':31} {'vin':'LSJA0000000000091','createTime':1546667597,'mileage':32} {'vin':'LSJA0000000000091','createTime':1546667598,'mileage':33} {'vin':'LSJA0000000000091','createTime':1546667599,'mileage':34} {'vin':'LSJA0000000000091','createTime':1546667600,'mileage':35} {'vin':'LSJA0000000000091','createTime':1546667601,'mileage':36} {'vin':'LSJA0000000000091','createTime':1546667602,'mileage':37} {'vin':'LSJA0000000000091','createTime':1546667603,'mileage':38} {'vin':'LSJA0000000000091','createTime':1546667604,'mileage':39} {'vin':'LSJA0000000000091','createTime':1546667605,'mileage':40} {'vin':'LSJA0000000000091','createTime':1546667606,'mileage':41} {'vin':'LSJA0000000000091','createTime':1546667607,'mileage':42} {'vin':'LSJA0000000000091','createTime':1546667608,'mileage':43} {'vin':'LSJA0000000000091','createTime':1546667609,'mileage':44} {'vin':'LSJA0000000000091','createTime':1546667610,'mileage':45} {'vin':'LSJA0000000000091','createTime':1546667611,'mileage':46} {'vin':'LSJA0000000000091','createTime':1546667612,'mileage':47} {'vin':'LSJA0000000000091','createTime':1546667613,'mileage':48} {'vin':'LSJA0000000000091','createTime':1546667614,'mileage':49} {'vin':'LSJA0000000000091','createTime':1546667615,'mileage':50} {'vin':'LSJA0000000000091','createTime':1546667616,'mileage':51} {'vin':'LSJA0000000000091','createTime':1546667617,'mileage':52} {'vin':'LSJA0000000000091','createTime':1546667618,'mileage':53} {'vin':'LSJA0000000000091','createTime':1546667619,'mileage':54} {'vin':'LSJA0000000000091','createTime':1546667620,'mileage':55} {'vin':'LSJA0000000000091','createTime':1546667621,'mileage':56} {'vin':'LSJA0000000000091','createTime':1546667622,'mileage':57} {'vin':'LSJA0000000000091','createTime':1546667623,'mileage':58} {'vin':'LSJA0000000000091','createTime':1546667624,'mileage':59} {'vin':'LSJA0000000000091','createTime':1546667745,'mileage':60} {'vin':'LSJA0000000000091','createTime':1546667746,'mileage':61} {'vin':'LSJA0000000000091','createTime':1546667747,'mileage':62} {'vin':'LSJA0000000000091','createTime':1546667748,'mileage':63} {'vin':'LSJA0000000000091','createTime':1546667749,'mileage':64} {'vin':'LSJA0000000000091','createTime':1546667750,'mileage':65} {'vin':'LSJA0000000000091','createTime':1546667751,'mileage':66} {'vin':'LSJA0000000000091','createTime':1546667752,'mileage':67} {'vin':'LSJA0000000000091','createTime':1546667753,'mileage':68} {'vin':'LSJA0000000000091','createTime':1546667754,'mileage':69} {'vin':'LSJA0000000000091','createTime':1546667755,'mileage':70} {'vin':'LSJA0000000000091','createTime':1546667756,'mileage':71} {'vin':'LSJA0000000000091','createTime':1546667757,'mileage':72} {'vin':'LSJA0000000000091','createTime':1546667758,'mileage':73} {'vin':'LSJA0000000000091','createTime':1546667759,'mileage':74} {'vin':'LSJA0000000000091','createTime':1546667760,'mileage':75} {'vin':'LSJA0000000000091','createTime':1546667761,'mileage':76} {'vin':'LSJA0000000000091','createTime':1546667762,'mileage':77} {'vin':'LSJA0000000000091','createTime':1546667763,'mileage':78} {'vin':'LSJA0000000000091','createTime':1546667764,'mileage':79} {'vin':'LSJA0000000000091','createTime':1546667765,'mileage':80} {'vin':'LSJA0000000000091','createTime':1546667766,'mileage':81} {'vin':'LSJA0000000000091','createTime':1546667767,'mileage':82} {'vin':'LSJA0000000000091','createTime':1546667768,'mileage':83} {'vin':'LSJA0000000000091','createTime':1546667769,'mileage':84} 开车的时候, 从一个地方到另一个地方, 会有很多行程。

2019年1月有闰秒吗？（剧透：没有） 今晚我们将在午夜时分从10倒数到0，并来到新的一年。 但是当我们倒数到0时，它会是2019年吗？如果有闰秒怎么办？ 快点 - 让我们检查IETF提供的闰秒列表。 不 - 它看起来不错，2017年1月1日是最后一个，并且该列表在2019年6月28日之前是 OK 的。 两年前，情况有所不同。 在“午夜”（UTC）时，它变成了2017年。 ~ $ raku -e 'say DateTime.new("2017-01-01T00:00:00Z").year' 2017 但实际上，从晚上（和年）的11:59:59开始已经过了两秒钟！ ~ $ raku -e 'say DateTime.new("2017-01-01T00:00:00Z") - DateTime.new("2016-12-31T23:59:59Z")' 2 在它变成2017年的那一刻，有一个闰秒。now返回的值表明了这一点。 虽然今晚会很好： ~ $ raku -e 'say DateTime.new("2019-01-01T00:00:00Z") - DateTime.new("2018-12-31T23:59:59Z")' 1 如果我们想要更明确，我们可以使用 posix 和 from-posix 将 DateTime 转换为 Instant 并再返回 - 后者将布尔值作为第二个参数来指示是否应该考虑闰秒。 my $date = DateTime.new("2017-01-01T00:00:00Z"); my $instant = Instant.from-posix: $date.posix, True; my $instant2 = Instant.from-posix: $date.posix; say $instant.DateTime.year; say $instant2.

在之前的文章中，我们探讨了如何使用带DStream抽象的Sparks Streaming API进行处理状态流。今天，我想和你一起探索Spark 2.2，它在Structured Streaming API下支持状态流。在这篇文章中，我们将看到API如何成熟和发展，看看两种方法（流与结构化流）之间的差异，并了解对API进行了哪些更改。我们将通过上一篇文章中的示例，并使其适应新的API来实现。 回顾 Spark Streaming 状态管理 如果你需要在Spark中使用状态流，你必须在两个抽象之间进行选择（直到Spark 2.2）。updateStateByKey 和 mapWithState后者或多或少是前者的改进（API和性能方面）版本（具有一些不同的语义）。为了利用微批次之间的状态，你提供了一个StateSpec函数，mapWithState对于到达当前微批次的每个键值对，将调用该函数。对于 mapWithState，主要优点是： 状态的初始RDD - 可以加载具有先前保存状态的RDD 超时 - 超时管理由Spark处理。你可以为所有键值对设置单个超时。 部分更新 - 仅迭代当前微批中“触摸”的键以进行更新 返回类型 - 你可以选择任何返回类型。 但事情并不总是完美的…… mapWithState 的痛点 mapWithState 比以前的 updateStateByKey API 有了很大的改进。但是去年在我使用它的过程中，体验过一些缺点： 检查点 为了确保Spark可以从失败的任务中恢复，它必须将数据 checkpoint 到分布式文件系统中，它可以在失败时从中恢复。使用 mapWithState 时，每个 executor 在内存中保存已累积的所有状态的 HashMap。在每个检查点，Spark 每次都会序列化整个状态。如果你在内存中持有很多状态，这可能会导致严重的处理延迟。例如，以下设置： 批处理间隔：4秒 检查点间隔：40秒（4秒批次x 10 常量 spark 系数） 每秒80,000条消息 消息大小：500B - 2KB 5 m4.2xlarge机器（8个vCPU，32GB RAM） 每台机器2个 executor executor 存储大小~7GB（每个） 将数据 checkpoint 到 HDFS 我经历了长达4小时的累积延迟，因为高负荷下的每个检查点在整个状态下花费30秒-1分钟，我们每4秒产生一次批次。我也看到人们在StackOverflow上对此感到困惑，因为实际上并不明白为什么有些批次比其他批次花费的时间要长得多。 如果你计划使用状态流来实现高吞吐量，则必须将此视为一个严重的警告。这个问题非常严重，以至于它让我找到了在Spark中使用内存状态的替代方法。但我们很快就会看到事情看起来很光明;） 在版本更新之间保存状态 软件是一个不断发展的过程，我们总是在改进，增强和实现新的功能要求。因此，我们需要能够从一个版本升级到另一个版本，最好不要影响现有数据。在内存数据中，这变得非常棘手。我们如何保持现状？我们如何确保从中断的地方继续？

欢迎来到第25天，这是Raku One-Liner Advent Calendar的最后一天！传统的 advent calendar 只有24个条目，我们今天的奖励帖子将专注于你可以在Raku高尔夫比赛中使用的一些提示和技巧。 有一个很棒的网站code-golf.io，你可以尝试解决一些问题，并将Raku移到最高分。我怀疑许多问题可以从这个One-Liner Advent Calendar 前几天所涵盖的技术中受益。 省略主题变量 如果在主题变量$_上调用方法，那么 Raku 实际上不需要变量名来理解你在说什么，所以，避免显式命名主题变量： $_.say for 1..10 使用范围进行循环 Perl 中的范围是表达循环细节的好东西：用几个字符，指定循环变量的初始和最终状态。倒装形式通常更短。 for 1..10 {.say} .say for 1..10 想想你是否可以从0开始计数，在这种情况下，可以使用脱字符来获得从0开始的范围。以下代码打印数字0到9： .say for ^10 在范围和序列之间进行选择 在循环中，序列可以与范围完全相同。选择可能取决于Golf软件是计数字节还是计数Unicode字符。在第一种情况下，两个点的范围优于三个点的范围。在第二种情况下，使用Unicode字符： .say for 1..10 .say for 1...10 .say for 1…10 当你需要向下计数时，序列是你的朋友，因为他们可以推断出循环计数器更改的方向： .say for 10…1 使用map而不是循环 在某些情况下，尤其是当你必须使用循环变量进行多个操作时，请尝试使用map迭代所有值： (^10).map: *.say 省略括号 与 Perl 5 不同，Raku 不会强制你在常规形式的条件检查中使用括号： if ($x > 0) {say $x;exit} if $x > 0 {say $x;exit} 有时，你也希望省略函数调用中的括号。 在声明数组或散列时，你都不需要括号。对于数组，使用引号构造：

总结个 🍉。

欢迎来到Raku One-Liner Advent Calendar的第24天！ 在前几天，我们正在读取文本文件，因此讨论 $*ARGFILES是合乎逻辑的，这是一个内置的动态变量，在处理多个输入文件时可能很方便。 如何读取命令行中传递的两个或多个文件？ $ raku work.pl a.txt b.txt 如果您需要将所有文件一起处理，就像它们是单个数据源一样，您可以要求我们今天的变量以单行方式完成工作： .say for $*ARGFILES.lines 在程序内部，您不必考虑循环文件; $*ARGFILES将自动为您做到这一点。 如果命令行中没有文件，则该变量将附加到 STDIN： $ cat a.txt b.txt | raku work.pl 确实很方便，不是吗？ 6.d和MAIN 如果你想在更大的程序中使用它，我也必须警告你。考虑以下程序： sub MAIN(*@files) { .say for $*ARGFILES.lines; } 在 Perl 6.d 中, $*ARGFILES 在MAIN子例程内部和它之外的工作方式不同。 这个程序将完全适用于 Raku.c，但不适用于Raku.d. 换句话说，在Rakudo Star中，包括版本2018.10，$*ARGFILES 在命令行中处理文件名，但从 Rakudo Star 2018.12 开始，如果在 MAIN 中使用它，它将始终连接到 $*IN。 这就是今天的 advent 文章的结束，几乎是今年整个日历的结束。不过，明天再来！

Thanks so much to Andrew Shitov, for writing so much nice articles, for allowing me to translate these posts to Chinese. If you are fluent in english, please read the source posts: Day 1 - https://raku.online/2018/12/01/generating-random-password/ Day 2 - https://raku.online/2018/12/02/grep-dividable-numbers/ Day 3 - https://raku.online/2018/12/03/generating-random-integers-in-perl-6/ Day 4 - https://raku.online/2018/12/04/working-with-big-numbers-in-perl-6/ Day 5 - https://raku.online/2018/12/05/what-is-the-date-today-in-raku/ Day 6 - https://raku.online/2018/12/06/palindrom-testing-in-perl-6/ Day 7 - https://raku.online/2018/12/07/the-joy-of-unicode-in-perl-6/ Day 8 - https://raku.online/2018/12/08/adding-up-even-fibonacci-numbers-in-perl-6/ Day 9 - https://raku.online/2018/12/09/more-on-x-in-perl-6/ Day 10 - https://raku.

欢迎来到Raku One-Liner Advent Calendar的第23天！今年年底是人们计算年度结果的时间，Raku 也可以帮助解决这个问题。 今天，我们将看到一个单行程序，用于计算表格列的总数。 这是文件中的一些示例数据： 100.20 303.50 150.25 130.44 1505.12 36.41 200.12 305.60 78.12 这里是为每列打印三个数字总和的代码： put [Z+] lines.map: *.words 该程序打印我们需要的数字： 430.76 2114.22 264.78 从昨天的帖子的更新中，我们知道，裸lines 和 $*IN.lines 是一样的，所以lines.map在输入流中迭代所有行。然后将每一行拆分成由空格分隔的单词子串。 解析输入数据的作业部分已完成。我们得到了许多与输入数据行相对应的序列。对于我们的示例文件，有如下内容： (("100.20", "303.50", "150.25").Seq, ("130.44", "1505.12", "36.41").Seq, ("200.12", "305.60", "78.12").Seq).Seq 现在，将每个序列的第一个元素，每个序列的第二个元素等相加起来。化简运算符和zip元运算符的组合只需要四个代码字符即可完成所有工作：[Z+]。 此时，我们有一个化简后的序列： （430.76,2114.22,264.78）.Seq 最后一个简单的步骤是使用 put 例程打印值。如果你昨天做了功课，你会知道say使用gist方法（在序列周围添加括号）来显示结果，而put只是使用Str方法打印值。 还有 让我们在脚本中添加一些字符来演示如何跳过第一列，例如，跳过包含月份名称的列： Jan 100.20 303.50 150.25 Feb 130.44 1505.12 36.41 Mar 200.12 305.60 78.12 您只需要创建一个切片并选择除第一列之外的所有列： put 'Total ', [Z+] lines.map: *.words[1..*] 如您所见，我们甚至不需要自己计算列数。1..* 范围可以完成这样的工作。 这就是今天的 advent 文章的结尾，所以明天再来吧！

你有一个可爱的Mojolicious应用程序，它是时候部署它了！ 但是……它不能在生产服务器上运行！到底是怎么回事？哦，不，您所依赖的模块与开发服务器上的版本不同。你能做什么？ 实际上，一些模块发展得很快（Hello Mojolicious！），这没有毛病，但可能会导致不兼容的变化。 还有一些可以在版本中解决或引入的错误，如果您的版本错误，则会遇到这些错误。 Cpanfile来救援 Cpanfile是一种用于描述Perl应用程序的CPAN依赖关系的格式。 有了cpanfile，我们可以列出我们需要的模块，但我们也可以强制模块的最小版本，它们的最大版本……或者说“我想要那个模块的确切版本”。 但我们也可以列出可选模块：您可以支持不同的数据库，但如果用户想要使用PostgreSQL，则不必安装与MySQL相关的模块。 这是一个例子cpanfile： # Do not ask for a specific version requires 'DateTime'; # Ask a specific version requires 'Plack', '== 1.0'; # Ask a minimal version requires 'Net::DNS', '>= 1.12'; # Or requires 'Net::DNS', '1.12'; # Ask a maximal version requires 'Locale::Maketext', '< 1.28'; # Give a range requires 'Mojolicious', '>= 7.0, < 8.0'; # Optional modules feature 'postgresql', 'PostgreSQL support' => sub { requires 'Mojo::Pg'; }; feature 'mysql', 'MySQL support' => sub { requires 'Mojo::mysql'; }; feature 'ldap', 'LDAP authentication support' => sub { requires 'Net::LDAP'; }; Cpanfile格式可以做更多（推荐模块，用于特定阶段（需求configure，test…），使用没有公布关于CPAN …模块），但是这是一个关于 Carton 的文章：我让你读cpanfile文档🙂

欢迎来到Raku One-Liner Advent Calendar的第22天！今天，我们将继续使用文件，今天的目标是创建一个单行程序以相反的顺序打印文本文件的行（如tail -r那样）。 第一个单行程序使用 STDIN 流完成工作： .say for $*IN.lines.reverse 运行程序为： $ raku reverse.pl < text.txt 如果要直接从Raku读取文件，请稍微修改程序以从命令行参数创建文件句柄： .say for @*ARGS[0].IO.open.lines.reverse 现在运行如下： $ raku reverse.pl text.txt 重要的是要记住，lines方法的默认行为是从最后的行序列中排除换行符（该方法返回Seq对象，而不是数组或列表）。它可能与你在使用Perl 5时习惯的方式相反。使用chomp是一种常见的做法。 在Raku中，lines方法根据IO::Handle对象的nl-in 属性中存储的值来拆分行。 您可以使用以下小脚本查看行分隔符的当前值： dd $_ for @*ARGS[0].IO.open.nl-in 这是您默认找到的内容： $["\n", "\r\n"] 有趣的是，你可以控制的 lines 的行为，并告诉 Perl 不排除换行符： @*ARGS[0].IO.open(chomp => False).lines.reverse.put 该 chomp 属性由默认设置为 True。您还可以更改默认分隔符： @*ARGS[0].IO.open( nl-in => "\r", chomp => False ).lines.reverse.put 请注意，如果没有 chomping，则不需要对行进行显式的for循环：在最后两个单行中，直接在序列对象上调用.put方法。在早期版本中，字符串不包含换行符，因此它们将打印为单个长行。 今天我将给你留一些小功课：告诉 say 和 put 之间的差异。 明天见！

Advent Calendar向您展示了许多使用Mojolicious的好方法，因为您已经安装了Mojo，所以除了Web处理之外，您还可以使用它。今天的配方使用模板渲染引擎进行Web响应之外的其他操作。 首先，处理一些字符串模板。这是使用v5.26中发布的波浪形heredoc语法从Mojo::Template中取出的一个示例 use Mojo::Template; my $mt = Mojo::Template->new; say $mt->render(<<~'EOF'); % use Time::Piece; <div> % my $now = localtime; Time: <%= $now->hms %> </div> EOF 带有前导百分号的行是Perl代码。其中一行加载模块，Time::Piece，另一行创建变量$now。该<%= %>插入值内联。您可以自己找出其他模板语法; 这些都在模块文档中。 您可以将其反转，以便值的来源来自模板外部。有时，这比在让表示层中有太多逻辑更可取： use v5.26; use Mojo::Template; my $mt = Mojo::Template->new; use Time::Piece; my $now = localtime; say $mt->render(<<~'EOF', $now->hms ); % my $time = shift; <div> Time: <%= $time %> </div> EOF 如上所示，默认情况下，参数以位置方式传递给模板。然而，通过名称而不是位置来描述变量更为自然。该vars属性打开您将哈希传递给模板的能力; 哈希键成为模板本身的变量名。 use v5.26; use Mojo::Template; my $mt = Mojo::Template->new->vars(1); use Time::Piece; my $now = localtime; say $mt->render(<<~'EOF', { time => $now->hms } ); <div> Time: <%= $time %> </div> EOF 从文件（或许多文件）中获取该模板同样容易。这是我以前使用Template Toolkit执行此操作的一种方式，这是一个非常精细且功能强大的模块，与以前一样好。但是，我已经在使用了Mojo的很多东西，它已经有了一个模板引擎。我可以减少依赖计数并专注于一种模板语言。

创建一个 Flink 应用程序的样板 mvn archetype:generate \ -DarchetypeGroupId=org.apache.flink \ -DarchetypeArtifactId=flink-quickstart-scala \ -DarchetypeVersion=1.6.0 \ -DarchetypeRepository=http://maven.aliyun.com/nexus/content/groups/public 开始 我有一个 Spark 应用程序, 我想把它转为 Flink 应用。在 Spark 应用里我读取 Kafka 中的消息为 DStream, 然后打印出来。我们使用 Flink 来做同样的事情。为了方便起见, 我们使用本地模式: 首先启动一个本地的 Flink 集群 /usr/local/Cellar/apache-flink/1.7.0/libexec/bin/start-cluster.sh 然后在 pom 文件中添加入口类(可选)： <mainClass>ohmysummer.BigdataApplication</mainClass> 然后启动这个应用：(如果是其它的 object, 使用 flink run -c className) flink run ./target/stream-word-count-1.0-SNAPSHOT.jar 打开浏览器, 输入网址: localhost:8081 查看 job 的运行状态, 即使你 Ctrl-C 了 flink 提交命令, 依然可以看见任务在运行。如果要取消, 在 UI 界面找到任务后点击 Cancel。 或查看输出： tail -f /usr/local/Cellar/apache-flink/1.7.0/libexec/log/flink-<whoami>-taskexecutor-0-<hostname>.out 停止集群： /usr/local/Cellar/apache-flink/1.7.0/libexec/bin/stop-cluster.sh

欢迎来到Raku One-Liner Advent Calendar的第21天！到本季即将结束的日历只剩下几天了，所以让我们在剩下的日子里收集尽可能多的东西，今天我们将几个文件合并到一个文件中 🙂 我们今天的目标是获取两个（或三个或更多）文件并逐行复制其内容。例如，我们想要合并两个日志文件，知道它们的所有行都相互对应。 文件 a.txt： 2018/12/20 11:16:13 2018/12/20 11:17:58 2018/12/20 11:19:18 2018/12/20 11:24:30 文件 b.txt： "/favicon.ico" failed (No such file) "/favicon.ico" failed (No such file) "/robots.txt" failed (No such file) "/robots.txt" failed (No such file) 我们的第一个单行程序阐述了这个想法： .say for [Z~] @*ARGS.map: *.IO.lines; 假设程序运行如下： $ raku merge.pl a.txt b.txt 对于命令行中的每个文件名（@*.ARGS.map），将创建一个IO::Path对象（.IO），并读取文件中的行（.lines）。 在两个文件的情况下，我们有两个序列，然后使用应用了连接中缀~的zip元运算符Z逐行连接。 在那一步之后，我们得到另一个序列，我们可以逐行打印（.say for）。 2018/12/20 11:16:13"/favicon.ico" failed (No such file) 2018/12/20 11:17:58"/favicon.ico" failed (No such file) 2018/12/20 11:19:18"/robots.txt" failed (No such file) 2018/12/20 11:24:30"/robots.

Dancer（及其他）PSGI 应用程序的作者很可能习惯于测试与Plack::Test，虽然这是一个崇敬的选择，这是很裸机。 在去年出现的过程中，我写了一篇关于Test::Mojo的文章，展示了许多简单易用（我敢说）有趣的方法，你可以用它来测试你的Mojolicious应用程序。如果你错过了，那就去看看吧。 我希望至少有一些人能够阅读并思考，“我很乐意使用它，但我不会使用Mojolicious！”; 好吧，你很幸运！只需要一点角色来弥补差距，你也可以使用Test::Mojo来测试你的PSGI应用程序！ 安装PSGI应用程序 Mojolicious本身不使用PSGI协议，因为它没有提供某些特性，而且某些异步操作也是必需的。也就是说，你可以使用Mojo::Server::PSGI在PSGI服务器上提供Mojolicious应用程序。当你的基于Mojolicious的应用程序检测到它已在PSGI服务器（例如plackup或Starman）下启动时，将自动使用此Mojolicious核心模块。 虽然在Mojo应用程序和PSGI服务器之间进行转换是核心功能，但相反，在PSGI应用程序和Mojolicious服务器（或应用程序，如你所见）之间进行转换可作为第三方模块使用。Mojolicious::Plugin::MountPSGI，顾名思义，可以将PSGI应用程序安装到基于Mojolicious的应用程序中。为此，它构建了一个新的，空的Mojolicious应用程序，在将任何请求转移到PSGI环境之前，将其转移到任何mount-ed应用程序。 使用Test::Mojo进行测试 一旦你能做到这一点，采用PSGI应用程序，用MountPSGI包装它，并将其设置为与Test::Mojo一起使用的应用程序是微不足道的。尽管如此，为了让它变得更加容易，在Test::Mojo::Role::PSGI中已经完成了所有这些工作。 与任何Mojolicious Role一样，我们可以使用应用角色with_roles创建（主要是匿名）子类。你可以使用快捷方式+代替Test::Mojo::Role::。 use Test::Mojo; my $class = Test::Mojo->with_roles('+PSGI'); 然后使用PSGI应用程序的路径实例化该角色，或者使用PSGI应用程序本身。 由于你使用的是角色，这些角色都与组合有关，因此你还可以应用可能在CPAN上找到的其他角色。 一个例子 作为一个例子，假设我们有一个简单的应用程序脚本（命名app.psgi），可以以不同的方式渲染 "hello world"或"hello $user" 。我将允许纯文本响应，JSON和模板化HTML（使用简单的模板来保持这种简洁）。 use Dancer2; set template => 'simple'; set views => '.'; any '/text' => sub { my $name = param('name') // 'world'; send_as plain => "hello $name"; }; any '/data' => sub { my $name = param('name') // 'world'; send_as JSON => { hello => $name }; }; any '/html' => sub { my $name = param('name') // 'world'; template 'hello' => { name => $name }; }; start; 而模板（hello.

欢迎来到 Raku One-Liner Advent Calendar 的第20天！到目前为止，我们创建了大约25种不同的单行程序，但从未讨论过Rakudo Raku编译器为我们提供的命令行选项。 -e 使用（Rakudo）Raku时的第一个选项是-e。它需要一个带有Raku单行程序的字符串并立即执行。 例如，打印当前Raku规范的版本： $ raku -e'$ * PERL.version.say' v6.c 注意不要使用Perl 5.10+样式的大写字母 -E，它与-e相同，但也激活诸如 say 之类的功能。在Raku中，选项始终为小写。 -n 此选项为每行输入数据重复运行代码。当您想要处理文件时，这非常方便。例如，这是一个单行程序，将行中的值相加并打印总和： raku -ne'say [+] .split(" ")' data.txt 如果data.txt文件包含以下内容： 10 20 30 40 1 2 3 4 5 6 7 8 然后单行程序的结果是： 100 10 26 这与你是否使用shell的输入重定向没有区别; 以下单行程序也有效： raku -ne'say [+] .split(" ")' < data.txt 确保将 e 选项放在选项列表中的最后一个（因此，不是raku -en'...'）或拆分选项：raku -n -e'...'。 -p 此选项类似于-n，但在每次迭代后打印主题变量。 以下单行程序反转文件中的行并将其打印到控制台： raku -npe'.=flip' data.txt 对于相同的输入文件，结果将如下所示： 04 03 02 01 4 3 2 1 8 7 6 5 请注意，您必须更新$_变量，因此键入.

把旧电脑上的 .ssh 目录拷贝到新电脑上, 继续使用之前的 ssh 免密码登录提示 「权限不够或需要输入密码」， 解决办法： chmod 755 ~/.ssh/ chmod 600 ~/.ssh/id_rsa ~/.ssh/id_rsa.pub chmod 644 ~/.ssh/known_hosts git 回滚 git log # 查看提交的 commit 历史 git reset –hard 8ff24a6803173208f3e606e32dfcf82db9ac84d8 git 瘦身 http://blog.mallol.cn/如何给git仓库瘦身删除大文件.html git pull 不输入密码 # vi .git/config [remote "origin"] url = https://Username:Password@github.com/myRepoDir/myRepo.git scp 复制文件 - 权限拒绝 ssh 可以登陆，但是 scp 复制文件时却不行： $ sudo scp -P 52113 ubuntu@123.206.136.59:/tmp/*.txt . Permission denied (publickey) 在 ~/.ssh/config 文件中添加： Host 123.206.136.59 IdentityFile ~/.ssh/ohmycloud

我的Yancy文档站点已经构建，具有自定义登录页面和POD查看器，我只需要部署该站点。我可以使用hypnotoad部署网站，Mojolicious的preforking服务器进行热部署，但这需要我有一台服务器并保持在线。如果我可以像Github一样部署静态网站，就像所有酷炫的人一样，那就好多了。 但要做到这一点，我需要把我的动态网站变成一个静态网站，这是不可能的！真的是吗？为什么我问我自己，我就是那个开辟道路的人：Mojolicious导出命令。 export 命令将一组路径作为输入，获取这些页面，并将结果写入目录。然后它会查看这些页面上的所有链接并编写这些页面。通过这种方式，它将整个 Mojolicious 网站导出为静态文件。 我需要做的就是能够使用 export 命令来安装它： $ cpanm Mojolicious::Command::export 一旦安装完毕，我们现在在我们的应用程序中有了 export 命令，我可以像任何其他 Mojolicious 命令一样使用它。 $ ./myapp.pl export 默认情况下，export命令尝试导出主页（/）并从那里递归工作。如果我的页面没有从其他地方链接，我应该（a）可能添加一些指向该页面的链接，但是（b）可以将它添加到要导出的页面列表中： $ ./myapp.pl export / /private 由于我在我的个人网站的目录下托管这个网站，我需要使用该--base选项重写所有内部链接到正确的路径，我可以使用该--to选项直接写入Web服务器的目录： $ ./myapp.pl export --base /yancy --to /var/www/preaction.me/yancy 而且，如果我愿意，我可以使用Mojolicious Config插件 来更改默认设置，包括要导出的页面，导出目录和 base URL。 最好的部分是export命令处理重定向。因此，当我们使用PODViewer插件并重定向到MetaCPAN时，页面会根据重定向的位置进行更新！ 将来，如果将此命令设置为插件，以便它可以具有用于自定义导出内容的钩子或对断开链接的附加检查，那将是很好的。如果有人有兴趣帮助完成这项工作，请告诉我，我可以帮助他们开始工作！ 现在，通过Yancy CMS，PODViewer插件和Mojolicious导出命令，我为Yancy提供了一个好看的文档网站！查看完整的应用程序。

brew install redis To have launchd start redis now and restart at login: brew services start redis Or, if you don't want/need a background service you can just run: redis-server /usr/local/etc/redis.conf 可以编辑 /usr/local/etc/redis.conf， 设置 redis 密码 requirepass 启动： # redis-server redis-server /usr/local/etc/redis.conf 127.0.0.1:6379> pfadd mykey a b c (integer) 1 127.0.0.1:6379> pfcount mykey (integer) 3 127.0.0.1:6379> pfadd mykey2 a b c c d (integer) 1 127.0.0.1:6379> pfcount mykey2 (integer) 4 127.

欢迎来到Raku One-Liner Advent Calendar的第19天！今天，我们将使用两种不同的方法计算π的值。这篇博文的目的是使用不同的方法来生成数字序列。 Pre-party 当然，在 Raku 中你不需要自己计算 π 的值，因为 Raku 给出了一些π和pi 形式的预定义常数，以及双倍的值τ和tau。 但为了演示 map 和序列的用法，让我们实现一种最简单的算法来计算 π： 这是检查答案的草案代码： my $pi = 1; my $sign = 1; for 1..10000 -> $k { $sign *= -1; $pi += $sign / ($k * 2 + 1); } say 4 * $pi; 第1部分 现在，让我们使用map来使答案紧凑一点。最好使公式更通用： 这是我们的第一个单行程序： say 4 * [+] (^1000).map({(-1) ** $_ / (2 * $_ + 1)}) 我希望你能理解这段代码里的所有东西。我们在今年的 Advent Calendar 的前几天介绍了该答案的不同部分，例如，在第11天的帖子中。 但是，我仍然想强调你需要-1周围的圆括号。如果输入-1 ** $_，则总是得到 -1，因为减号前缀应用于取幂的结果。所以正确的代码是(-1) ** $_。

在$work 中，我们使用 Dancer 构建了一个 API，用于生成 PDF 文档和 XML 文件。此 API 是保险登记系统的重要组成部分：生成PDF以立即在Web浏览器中传送给客户端，并且XML一旦可用就会立即传送给运营商。由于XML通常需要花费大量时间来生成，因此在后台生成作业，以免在较长时间内占用应用程序服务器。完成后，开发了一个自行开发的流程管理系统，通过fork() 进程工作，跟踪其 pid，并希望我们以后可以成功地获得完成的过程。 这种方法存在一些问题： 很脆弱 它没有规模 作为开发人员搞砸某事太容易了 在2019年，我们不得不加大承担更大的工作量。目前的解决方案根本无法处理我们预计需要处理的工作量。直到遇见了 Minion。 **注意：**本文中使用的技术在 Dancer或 Dancer2同样适用。 为何选择Minion？ 我们研究了Minion的几种替代品，包括beanstalkd和celeryd。然而，使用其中任何一个意味着涉及我们已经过度征用的基础设施团队; 使用Minion允许我们使用我的团队已经拥有的专业知识，而不必向其他人提供帮助。从开发的角度来看，使用Perl开发的产品为我们提供了最快的实施时间。 扩展我们现有的设置几乎是不可能的。处理我们 folk 的进程所消耗的资源不仅难以处理，而且不可能在多个服务器上运行作业。从Minion开始也给了我们一个急需的机会来清理一些需要重构的代码。通过最少量的工作，我们能够清理我们的XML渲染代码并使其在Minion中运行。通过此清理，我们可以更轻松地获取有关XML渲染作业消耗了多少内存和CPU的信息。这些信息对于我们规划未来容量至关重要。 进入 Minion 既然我们是一个 Dancer 商店，而不是 Mojolicious，你从Mojolicious获得的与Minion工作的很多东西都不适合我们。鉴于我们还在商业模式中共享一些基于Minion的代码，我们必须围绕Minion构建一些自己的管道： package MyJob::JobQueue; use Moose; use Minion; use MyJob::Models::FooBar; with 'MyJob::Roles::ConfigReader'; has 'runner' => ( is => 'ro', isa => 'Minion', lazy => 1, default => sub( $self ) { $ENV{ MOJO_PUBSUB_EXPERIMENTAL } = 1; Minion->new( mysql => MyJob::DBConnectionManager->new->get_connection_uri({ db_type => 'feeds', io_type => 'rw', })); }, ); 我们在Minion周围包含了一个简单的Moose类，以便使用我们想要的额外功能轻松添加到任何类或Dancer应用程序。

第18天：对POD的看法 为了让Yancy拥有一个好的文档站点，实际上它需要渲染文档。要在Mojolicious中渲染Perl文档 ，我可以使用 PODViewer 插件（现已弃用的PODRenderer 插件的一个分支 ）。 将PODViewer添加到现有站点很容易！ use Mojolicious::Lite; plugin 'PODViewer'; app->start; 现在，当我访问http://127.0.0.1:3000/perldoc 时我看到了Mojolicious的 POD::Guides。这很棒，但是这是 Yancy 的文档站点，而不是 Mojolicious 的。让我们调整一些配置来使 Yancy 成为默认模块，并且只允许查看Yancy 模块（尝试查看另一个模块会将用户重定向到MetaCPAN）。 use Mojolicious::Lite; plugin 'PODViewer', { default_module => 'Yancy', allow_modules => [qw( Yancy Mojolicious::Plugin::Yancy )], }; app->start; 在那里，现在Yancy文档显示在首页。 最后，让它看起来更好一些：文档肯定需要使用默认的站点布局，而一些额外的CSS也会使文档看起来好看多了！ use Mojolicious::Lite; plugin 'PODViewer', { default_module => 'Yancy', allow_modules => [qw( Yancy Mojolicious::Plugin::Yancy )], layout => 'default', }; app->start; __DATA__ @@ layouts/default.html.ep <!DOCTYPE html> <html> <head> <link rel="stylesheet" href="/yancy/bootstrap.css"> <style> h1 { font-size: 2.

欢迎来到Raku One-Liner Advent Calendar的第18天！今天，将会有一个真正的单行程序，在某种意义上说，你要作为一个运维从终端中运行它。 我们的任务是重命名命令行参数中传递的所有文件，并以偏爱的格式提供文件序列号。以下是命令行的示例： $ raku rename.pl * .jpg img_0000.jpg 在此示例中，当前目录中的所有图像文件将重命名为img_0001.jpg，img_0002.jpg等。 这是 Raku 中可能的解决方案： @*ARGS[0..*-2].sort.map: *.Str.IO.rename(++@*ARGS[*-1]) 预定义的动态变量@*ARGS包含命令行中的参数。在上面的示例中，shell 将 *.jpg 掩码展开到文件列表中，因此数组中包含所有这些掩码。最后一个元素是重命名样本 img_0000.jpg。 请注意，与 Perl 5 不同，该变量称为 ARGS，而不是 ARGV。 为了遍历所有文件（并使用文件掩码跳过最后一个文件项），我们正在进行@*ARGS的切片。0..*-2 结构创建了索引范围以接收除了最后一个元素的所有元素。 然后对列表进行排序（原始的@*ARGS数组保持不变），我们使用map 方法迭代文件名 。 map 主体包含一个WhateveCode块 ; 它接受当前值的字符串表示形式，从中生成一个IO::Path对象，并调用 rename 方法。请注意，IO方法创建IO::Path类的对象; 而裸 IO是Raku对象系统层次结构中的一个角色。 最后，增量运算符++更改重命名样本（保存在@*ARGS 中的最后一个, 即 *-1 元素）。当操作符应用于字符串时，它会增加它的数字部分，因此我们得到 img_0001.jpg，img_0002.jpg等。 我希望 Raku 不会被称为 star-noise 语言 🙂 。尽管如此，明天还会有另一篇关于 Raku 的短篇小文！

您刚刚阅读了 如何在浏览器中减轻体重， 并希望了解减少Mojo应用程序的效果。该过程的一部分是阻止浏览器请求几乎不变的文件。我花了一个充满咖啡因的下午试图用Mojolicious做到这一点。我一直在’围绕房子，扰乱警报我直到最后才找到答案，有点像你最喜欢的圣诞节动画特别与一个小的林地生物叙述“Gruffalo的HTTP标题”。 儿童故事 我们心爱的小型林地生物需要显示一个森林日历，其中包含从数据库中提取的森林事件。Perl 可以获取事件数据并将其打包为JSON feed 流。Mojolicious 可以为每个用户准备具有正确 JSON feed 流的网页。使用一些JavaScript 库来显示 Web 日历，所有这些都可以在森林中使用。 除 JavaScript 库之外的所有内容都是轻量级的。每个人都知道, 如果不是每次都去下载 Javascript, 那么页面重载的速度会更快。那些库几个月都不会改变！如果只有客户端浏览器知道它可以使用上次下载的文件。 当然，秘诀是设置Cache-ControlHTTP 头的字段，但是怎样设置呢？ 首先，有一匹马 …… 每个使用 Apache 的人都会考虑使用 mod_expires ，这看起来很容易，除了 Apache 没有被用来为网页提供服务。 …但是马儿提到了一些甜蜜的 Cache-Control指令可以咀嚼，同时继续在早先下载的一些HTTP缓存页面上吃草 。小动物继续前进。 …… 然后有一只蟾蜍 森林生物使用 Mojolicious附带的 Hypnotoad Web服务器来为其页面提供服务。他们发现它非常适合他们的树栖生产环境。 它可以设置HTTP标头将其转换为反向代理，但是流行的设置是将Hypnotoad 置于 Nginx 或Apache/mod_proxy后面。那些服务器应该让你玩Expires标题。但是蟾蜍没有完全具备这种特殊的啮齿动物所需要的东西。 抛开 - 不，我没有提到 Plack。也许如果我今年好，圣诞老人会 告诉我 应该如何使用它。可能与某事有关 Plack::Response->header('Expires' => 'Tue, 25 Dec 2018 07:28:00 GMT'); 但我不知道，我的叙述者也没有。 …… 然后有一个独角兽 嗯，这很容易。只需使用标准的 Mojo::Headers 模块来设置Expires标头。 可是等等！这为一个不是那么大的页面设置了它。我们毛茸茸的朋友只想阻止 JavaScript 文件每次都重新加载，这会破坏 Sciuridae 的移动体验。Hmmmm。

Web应用程序经常需要向其用户发送电子邮件，例如收据或密码重置链接。Dancer2 的Email插件通过提供email关键字和健全的默认配置简化了此任务。 所以不可避免的“Hello world”示例如下所示： email { from => 'foo@perl.dance', to => 'bar@perl.dance', subject => 'Hello world', text => 'Welcome to the dancefloor!', }; 更常见的情况是使用Web应用程序中的模板并将其转换为HTML电子邮件。 你不是使用template关键字将HTML从路由返回到浏览器，而是生成具有特定布局的 HTML，存储在变量中并发送电子邮件。 post '/welcome' => { my $html = template $template, $tokens, { layout => 'email' }; email { from => 'foo@perl.dance', to => 'bar@perl.dance', subject => 'Welcome to the dancefloor!', type => 'html', body => $html, } redirect '/home'; } UTF-8 如果内容和邮件标题可能包含UTF-8字符，请注意对其进行编码。如果你不这样做，它甚至可能在你的电子邮件客户端中看起来很好，但不一定适用于其他客户端。 email { from => 'foo@perl.

今年，我决定Yancy需要一个网站。而不是建立一个像Statocles这样的静态网站生成器的网站，这些天如此受欢迎，我决定做一些疯狂和不可预测的事情：一个动态网站！幸运的是，我有一个完美的项目，可以轻松建立一个动态的网站：Yancy！ 任何动态网站的关键部分都是数据库。由于我只想编写 Markdown 并渲染 HTML，因此我的架构非常简单：存储页面路径的位置，存储页面 Markdown 以进行编辑的位置以及放置渲染 HTML 的位置。我建立了一个 SQLite 数据库并使用Mojo::SQLite::Migrations构建了pages 表 。 #!/usr/bin/env perl use Mojolicious::Lite; use Mojo::SQLite; helper db => sub { state $db = Mojo::SQLite->new( 'sqlite:' . app->home->child( 'docs.db' ) ); return $db; }; app->db->auto_migrate(1)->migrations->from_data(); # Start the app. Must be the code of the script app->start; __DATA__ @@ migrations -- 1 up CREATE TABLE pages ( id INTEGER PRIMARY KEY AUTOINCREMENT, path VARCHAR UNIQUE NOT NULL, markdown TEXT, html TEXT ); 准备好我们的数据库表后，我需要一种编辑页面的方法。Yancy 的内置编辑器带有marked.

欢迎来到 Raku One-Liner Advent Calendar 的第17天！今天，我们将有两个单行程序，它们都产生一些素数。 第1部分 首先，让我们解决欧拉项目的问题7，你需要打印第 10001 个素数（第一个素数是2）。 Raku 擅长素数，因为它有一个 Int 类的内置方法，is-prime。 有几种方法可以生成素数。对于单行程序，最简单的就是最好的, 效率也最低, 方法就是测试每个数字。 say ((1..*).grep: *.is-prime)[10000] 计算结果大约需要半分钟，但代码很短。总有一天，我们将使用所谓的埃拉托色尼筛来解决这个问题，它应该更快，但可能需要更多的代码。 第2部分 在这个 advent 文章的第二部分，让我们娱乐一下并解决 code-golf.io 网站上的相应问题。我们需要打印100以下的所有素数。 我的答案需要22个字符，如下所示： .is-prime&&.say for ^Ⅽ 在 Raku 中没有更短答案了，而在J中，他们做到了只使用11个字符。在 Raku 中，方法名已经消耗了八个字符。我相信，要赢得所有的高尔夫比赛，你需要一种名称很短的特殊语言（J就是）和一组内置例程来生成素数，或斐波那契或任何其他数字序列的列表。它还应该强烈利用Unicode字符空间。 在我们的 Raku 示例中，还有一个 Unicode 字符，Ⅽ。这不是简单的 C，拉丁字母的第三个字母，而是 Unicode 字符 ROMAN NUMERAL ONE HUNDRED（原本是拉丁字母的第三个字母，当然）。使用此符号，我们可以在答案中节省两个字符。 && 技巧是可能的，因为如果第一个操作数是 False，Perl 不会执行布尔表达式的第二部分。请注意，您不能在此处使用单个 ＆。完整的非优化版本的代码需要额外的空格，如下所示： .say if .is-prime for ^100 这就是今天 Raku 之旅的终点，明天见！

欢迎来到 Raku One-Liner Advent Calendar 的第16天！今天，我们将解决一个简单的问题，并找到平面上两点之间的距离。 这是一个有助于制定任务的插图。我们的目标是找到A点和B点之间的距离。 为了使答案更加透明和易于检查，我选择了AB线段，使得它是一个带有边长为3和边长为4的直角三角形的斜边。在这种情况下，第三边的长度将是5。 所以，这是答案： say abs(5.5+2i - (1.5+5i)) 代码使用了复数，一旦你移动到一个复平面，你就可以从平面上两点之间的距离等于这两个数相减的绝对值这一事实中获益。 在这种情况下，其中一个点是复平面上的点 5.5+2i，第二个点是 1.5+5i。在 Raku 中，您可以像在数学中一样编写复数。 如果没有复数的内置支持，你必须明确地使用毕达哥拉斯定理： say sqrt((5.5 - 1.5)² + (2 - 5)²) 家庭作业。修改 Rakudo 的 grammar 以允许以下代码： say √((5.5 - 1.5)² + (2 - 5)²) 这就是今天的一切。明天再来看看另外一个 Raku 单行或两个！

前所未有的表现力 Perl 5 正则表达式的强大功能使该语言成为分析文本数据的首选工具。从那时起，许多其他编程语言都复制了 Perl 的正则表达式，这部分地削弱了 Perl 在该领域中优于其他语言的优势。 从 Perl 5 派生的新 Raku 语言创建了一个新的文本匹配模型，该模型源自正则表达式，但功能强大且富有表现力，并且从常规正则表达式中删除它已决定给它们一个新名称，正则表达式。 不仅 Perl 的正则表达式6的机理是它大大高于现有的所有系统的正则表达式更强大，但它被设计成正则表达式相结合，构建 Grammar 语境，也就是说能够实现词汇和语法分析（系统词法和解析）到更复杂的数据，例如 HTML 文本，XML，XHTML，JSON，YAML，其中，不包括简单的情形，都超出正则表达式的范围。这些 Grammar 甚至可以分析所有级别的计算机程序。Raku 程序本身是用 Raku 自己编写的 Grammar 编译的。 虽然它们远不是 Raku 的唯一创新，但我们相信 Raku 正则表达式和 Grammar 将至少与 Perl 的正则表达式一样彻底改变计算机语言，甚至可能更多。他们现在还没有时间这样做。 本教程的讨论在 Perl 论坛上公开，地址如下： 评论 笔者 劳伦特罗森菲尔德 文章 发表于 2015年11月6日 - 更新于2018年 11月4日 PDF, 版离线版, ePub, Azw, Mobi 社交链接 1. 正则表达式入门 在正则表达式（或正则表达式）从数学和将字符串通常已知的形式语言的计算机科学理论概念模式（或模式）来描述一个整体（成品或不）字符串由模式定义的共同特征，根据预定义的语法并且不考虑上下文。的图案（匹配图案匹配）是应用这些模式到文本的样本，以试图找到对应于这些模式的文本片段的过程。 在你的计算机上安装Raku 如果你想使用 Raku，我们建议你在此地址下载 Rakudo Star。有关安装的更多信息，请参阅Perl 5到Raku教程的第一部分- 第1部分：语言基础知识。 在我们更新本文档（2018年10月）时，建议你选择 MoarVM 虚拟机。 1-1 与文件搜索类比 要在目录中搜索名称以字母 “a” 开头且扩展名为 .

Mojo

欢迎来到 Raku One-Liner Advent Calendar 的第15天！今天，将有两个单行程序，它们都产生斐波纳契数。 是的，最有可能的是，你从未在实际代码中使用过这样的数字，而且，很可能，你用它们解决了许多教育问题。然而，今天，让我们解决欧拉项目的问题25，并尝试在 Code-Golf.io 站点上寻找最短的 Fibonacci 问题解决方案。 Pre-party 我们如何在 Raku 中形成 Fibonacci 序列呢？使用序列运算符 ...: 0, 1, * + * ... * 如果你想在代码中有一些奇特的味道，你可以用 Inf 或 ∞ 替换最后的那个星号。在任何情况下，结果都是 Seq 类型的惰性序列。 Raku 不会立即计算它（并且它不能，因为右边缘是无限的）。 第1部分 第一个任务是找到第一个 Fibonacci 数的索引，该数字有 1000 个数字。当然，你可以循环上面创建的序列并自己跟踪索引。但是在 Raku 中，有一个选项可以修改 grep 例程簇，并要求它返回匹配项的索引而不是项本​​身。 另外，我们将使用更合适的方法 first 代替 grep。如果我们用 k 键调用该方法，它将返回第一个匹配项或其索引。仅仅提到键就足够了，真的不需要值。 say (0, 1, * + * ... *).first(*.chars >= 1000, :k) 该程序打印一个整数，这是给定问题的正确答案。 第2部分 现在让我们解决一个高尔夫任务并打印前30个斐波那契数字，用单行程序。这次，我们必须在代码中使用尽可能少的字符。 第一种方法相当冗长（即使使用 ^31 代替 0..30，它需要 33 个字符）：

Parse 入门 Rebol 语言最大的特色之一就是它的解析引擎，简称为 Parse。这是卡尔·萨森拉斯（Carl Sassenrath）的一个惊人设计，在过去的15年里，所有 Rebol 用户都免于使用臭名昭著的无法维护的 regexp 的痛苦。现在，Parse 也可用于 Red 用户，而且是增强版！ 那么，简而言之，什么是 Parse？它是一个嵌入式 DSL（我们称之为 Rebol 世界中的“方言”），用于使用语法规则解析输入序列。 Parse 方言是 TDPL 家族的增强成员。 Parse 的常见用法是检查，验证，提取，修改输入数据，甚至实现嵌入式和外部 DSL。 parse 函数调用语法很简单： parse <input> <rules> <input>: 任意系列值 (字符串, 文件, 块儿, 路径, ...) <rules>: 块儿! 具有有效 Parse 方言内容的值 这里有几个例子，即使你不懂 Red 和解析方言，你仍然可以“获得”它们中的大多数，不像正则表达式。您可以将它们直接复制/粘贴到 Red 控制台中。 使用语法规则进行字符串或块输入验证的一些简单示例： parse "a plane" [["a" | "the"] space "plane"] ;-- rule 中的空格需要用 space 显式声明 parse "the car" [["a" | "the"] space ["plane" | "car"]] parse "123" ["1" "2" ["4" | "3"]] ;-- rule 中的空格默认被忽略 parse "abbccc" ["a" 2 "b" 3 "c"] ;-- 一个 a, 俩个 b, 三个 c parse "aaabbb" [copy letters some "a" (n: length?

随着最新版本的Mojolicious，Mojo::DOM获得了很多力量，我很兴奋尝试，但没有时间。最近，我在实际工作中遇到了一个问题（在工程，采购和施工，或简称EPC），我在很短的时间内用Mojo::DOM（和Mojolicious的其他部分）解决了 - 包括学习如何使用Mojo::DOM，我以前从未做过。 任务 - 简单，但乏味 3D模型和绘图是很棒的工具，但实际上并不是那么完美。建筑公差就是它们的原因，我们公司非常依赖激光扫描，我们在那里进入项目现场并创建一个竣工条件的点云。这会生成数百个需要花费数小时才能处理的文件 - 最近的一个项目有超过30亿个人点。这些对于我们所做的每个项目的工程，建模和构造都至关重要。 问题是当我们的3D建模软件（Tekla Structures）处理点云时，它会将每个文件名从人类可读的东西（例如Pipe Rack Area1）更改为散列，如2e9d52829f973c5b98f60935d8a9fa2b。当一个项目可能有几十个区域时，这不是非常用户友好，而您实际上只想一次加载一个或两个（平均面积为15gb！）。 点云提供的关键信息太难或太昂贵，无法直接建模 幸运的是，Tekla使用了许多任何人都可以编辑的标准文件格式 - 包括使用XML来描述它已处理的每个点云。当然，我可以手工编辑它们来更改名称，但是每次扫描和每个项目都必须这样做 - 这不是一个好的解决方案。 方便的是，XML文件包含哈希名称和原始文件名 - 所以我知道我可以使用Mojo::DOM来解析XML并将点云重命名为人类可读。这个简单的任务就是Mojolicious如何用相对较短的代码完成大量工作的完美例子。结果如下： #!/usr/bin/perl use Mojo::Base -base; use Mojo::Util qw(getopt); use Mojo::File; use Mojo::DOM; getopt 'p|path=s' => \my $path; sub main { # look in xml elements for laserscans that have hashes for names my $file = Mojo::File->new($path, 'pointclouds.xml'); my $dom = Mojo::DOM->new($file->slurp); # if 'Hash' is populated, rename_files(); otherwise ignore for my $e ($dom->find('PointCloudData')->each) { $e->{Folder} = rename_files($e) and $e->{Hash} = '' if $e->{Hash}; } # save xml file so we don't try to rename the pointclouds again $file->spurt($dom); } sub rename_files { # rename pointcloud folder and database file my $e = shift; my $newname = $e->{Folder} =~ s/$e->{Hash}/$e->{Name}/r; say "renaming: $e->{Folder} to:\n$newname"; rename $e->{Folder}, $newname || die $!

欢迎来到 Raku One-Liner Advent Calendar 的第14天！今天，我们正在提出我们昨天解决的问题的另一个答案。任务是计算二十世纪第一个月的所有星期日。 昨天，我们只扫描了整个世纪的所有日子，选择星期日（ .day-of-week == 7）并且是本月的第一天（.day == 1）。 可以制作更有效的算法。由于我们只对本月的第一天感兴趣，因此无需在100年内扫描所有36525天，而只需要扫描100天，这是1901年到2000年之间每个月的第一天。 因此，我们需要两个嵌套循环：年和月。我们需要两个 for 吗？不必要。我们使用 X 运算符;我们从以前的出版帖子中熟悉它。 这是我们今天的单行： (gather for 1901..2000 X 1..12 { take Date.new(|@_, 1) }).grep(*.day-of-week == 7).elems.say; for 1901..2000 X 1..12 循环在遍历二十世纪的每个月份。对于每个循环变量，我们通过调用具有三个参数的构造函数来创建 Date 对象。 请注意，在循环内部，您可以同时使用 $_[0] 和 $_[1]，以及 @_[0] 和 @_[1]。在第一种情况下，$_ 变量包含两个元素的列表，而在第二种情况下，它是一个数组 @_。如果您只是使用点来调用主题（默认）变量上的方法，则可以实现最短的代码：.[0] 和 .[1]。 您可以键入 Date.new(.[0], .[1], 1) 代替 Date.new(|@_, 1)。 |@_ 语法用于展开数组，否则 Raku 会认为您将数组作为第一个参数传递。 在 gather-take 对儿的帮助下，将这些月份的所有第一天收集到序列中。 最后一步是昨天的 grep，但这次我们只需要选择星期日，所以单个 *.day-of-week == 7 条件就足够了。 elems 方法的调用返回列表中元素的数量，即我们正在寻找的星期日数量。因此，请用 say 打印出来。

pom 依赖： <!-- https://mvnrepository.com/artifact/com.hortonworks/shc-core --> <dependency> <groupId>com.hortonworks</groupId> <artifactId>shc-core</artifactId> <version>1.1.1-2.1-s_2.11</version> </dependency> 部分代码： package xxxx.xxxx import org.apache.log4j.{Level, Logger} import org.apache.spark.sql.execution.datasources.hbase._ import org.apache.spark.sql.functions._ import org.apache.spark.sql.{DataFrame, SparkSession} /** * 提交方式 * spark2-submit --name HBASE-CONNECTOR --files /etc/hbase/conf/hbase-site.xml --class xxx.xxxx --master yarn --deploy-mode cluster --driver-memory 2g --driver-cores 2 --executor-memory 2g --executor-cores 1 --num-executors 2 wmanxiety-1.0-SNAPSHOT.jar --day 20180810 --repartition 500 --interval 7 */ object MileageAnxiety { def cat = s"""{ |"table":{"namespace":"default", "name":"trip_signal", "tableCoder":"PrimitiveType"}, |"rowkey":"key", |"columns":{ |"rowkey" :{"cf":"rowkey", "col":"key", "type":"string"}, |"vin" :{"cf":"info", "col":"vin", "type":"string"}, |"tripStatus" :{"cf":"info", "col":"tripStatus", "type":"string"}, |"tripStartTime":{"cf":"info", "col":"tripStartTime", "type":"string"}, |"tripEndTime" :{"cf":"info", "col":"tripEndTime", "type":"string"}, |"tripDistance" :{"cf":"info", "col":"tripDistance", "type":"string"}, |"startSoc" :{"cf":"info", "col":"startSoc", "type":"string"}, |"endSoc" :{"cf":"info", "col":"endSoc", "type":"string"}, |"maxSpeed" :{"cf":"info", "col":"maxSpeed", "type":"string"}, |"startMileage" :{"cf":"info", "col":"startMileage", "type":"string"}, |"coordinate" :{"cf":"info", "col":"coordinate", "type":"string"} |} |}""".

在我之前的 Advent 文章中，我创建了高阶 promise并向您展示了如何使用它们。我没有告诉你它们如何工作的魔力。现在我将从另一个方向开发另一个例子。 有时候我希望 Mojo::File 的行为有点不同。我经常有一个路径，我想只将基本名称与不同的目录结合起来。我最终为两者制作 Mojo::File 对象，然后使用目录对象来获得我想要的东西： use Mojo::File qw(path); my $path = Mojo::File->new( '/Users/brian/bin/interesting.txt' ); my $dir = Mojo::File->new( '/usr/local/bin' ); my $new_path = $dir->child( $path->basename ); say $new_path; # /usr/local/bin/interesting.txt 那很烦人。我不喜欢它需要这么多步骤。我想要一些方法。我宁愿能够像这样编写它，我从有趣的文件开始并继续处理它而不是切换到其他对象： use Mojo::File qw(path); my $new_path = Mojo::File ->new( '/Users/brian/bin/interesting.txt' ) ->rebase( '/usr/local/bin' ); # this isn't a method say $new_path; # /usr/local/bin/interesting.txt 我可以通过各种 Perl 技巧通过猴子补丁或子类化将此方法添加到 Mojo::File。但是，像往常一样，Mojolicious 期待我的愿望并提供一种方法来做到这一点。我可以添加一个角色， 当我跳进它时，你可以自己阅读角色。首先，我创建一个代表我的角色的类。我定义了我想要的方法。我使用我想要影响的包的名称，添加 ::Role::，然后我想使用的名称;它的小写并不重要。 Mojo::Base 设置了当我导入 -role 时所需的一切： package Mojo::File::Role::rebase { use Mojo::Base qw(-role -signatures); sub rebase ($file, $dir) { $file->new( $dir, $file->basename ) } } 我在我想要影响的类上使用 with_roles 来应用我的新功能。因为我使用命名约定通过在它前面加上目标类（Mojo::File），然后是 ::Role::，然后是我想要的短名称。当我应用这个时，我可以省去大部分包裹名称，并使用前面加号的短名称：

欢迎来到 Raku One-Liner Advent Calendar 的第13天！今天的单行将很长，最好用两行编写，但它会显示 Raku 的 Date 对象的一个​​非常好的功能：它可以很容易地在一个范围内使用。 今天，我们正在解决欧拉项目的问题19。从本质上讲，任务是计算在1901年1月1日至2000年12月31日期间的星期日，这是在第一个月。 Raku 中的 Date 对象实现了 succ 和 prec 方法，它们递增和递减日期。也可以使用两个日期作为范围的边界： say ( Date.new(year => 1901) ..^ Date.new(year => 2001) ).grep({.day == 1 && .day-of-week == 7}).elems 在这里评论一下。 首先，使用单个命名参数（year）创建两个 Date 对象。这是可能的，因为构造函数的签名具有月和日的默认值： multi method new( Date: Int:D() :$year!, Int:D() :$month = 1, Int:D() :$day = 1, :&formatter, *%_) { . . . } 因此，创建1月1日的日期很容易，但是在一年的最后一天你不能这样做。但是 Raku 有一个很好的范围运算符 ..^，它排除了右边界并允许我们节省相当多的字符（而我们还没有玩 Raku Golf :-)。 具有日期的所有显式部分的较长版本将如下所示： say ( Date.

我使用的是 CDH5.13.2， 所以 pom 文件依赖为： <dependency> <groupId>org.apache.phoenix</groupId> <artifactId>phoenix-spark</artifactId> <version>4.14.0-cdh5.13.2</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.phoenix</groupId> <artifactId>phoenix-core</artifactId> <version>4.14.0-cdh5.13.2</version> </dependency> 这样读取： val df = sqlContext.read.format("jdbc") .option("driver", "org.apache.phoenix.jdbc.PhoenixDriver") .option("url", "jdbc:phoenix:localhost:2181") .option("dbtable", "US_POPULATION") .load() 或者 val df = sqlContext.load( "jdbc", Map("zkUrl" -> "localhost:2181", "url" -> "jdbc:phoenix:localhost:2181", "dbtable" -> "US_POPULATION", "driver" -> "org.apache.phoenix.jdbc.PhoenixDriver") )

在 $work，我们使用 Dancer 构建了一个 API，用于生成 PDF 文档和 XML 文件。此 API 是保险登记系统的重要组成部分：生成 PDF 以立即在 Web 浏览器中传送给客户端，并且 XML 一旦可用就会立即传送给运营商。由于XML 通常需要花费大量时间来生成，因此在后台生成作业，以免在较长时间内占用应用程序服务器。完成后，开发了一个自行开发的流程管理系统，并通过 fork() 进程，跟踪其 pid，并希望我们以后可以成功获得完成的流程。 这种方法存在一些问题： - 它很脆弱 - 它不能扩展 - 作为开发人员搞砸某事太容易了 在2019年，我们不得不加大承担更大的工作量。目前的解决方案根本无法处理我们预计需要处理的工作量。z直到遇见 Minion。 注意：本文中使用的技术与 Dancer 或 Dancer2 同样适用。 为何选择 Minion？ 我们研究了 Minion 的几种替代品，包括 beanstalkd 和 celeryd。然而，使用其中任何一个意味着涉及我们已经过度征税的基础设施团队;使用 Minion 允许我们使用我的团队已经拥有的专业知识，而不必向其他人提供帮助。从开发的角度来看，使用 Perl 开发的产品为我们提供了最快的实施时间。 扩展我们现有的设置几乎是不可能的。处理我们分叉的进程所消耗的资源不仅难以处理，而且不可能在多个服务器上运行作业。从 Minion 开始也给了我们一个急需的机会来清理一些需要重构的代码。通过最少量的工作，我们能够清理我们的 XML 呈现代码并使其在 Minion 中运行。通过此清理，我们可以更轻松地获取有关XML呈现作业消耗了多少内存和 CPU 的信息。这些信息对于我们规划未来容量至关重要。 进入 Minion 由于我们是一个舞蹈工作室，而不是 Mojolicious，因此我们可以从 Mojolicious 那里获得很多与 Minion 合作的便利。鉴于我们还在商业模式中共享一些基于 Minion 的代码，我们必须围绕 Minion 构建一些自己的管道： package MyJob::JobQueue; use Moose; use Minion; use MyJob::Models::FooBar; with 'MyJob::Roles::ConfigReader'; has 'runner' => ( is => 'ro', isa => 'Minion', lazy => 1, default => sub( $self ) { $ENV{ MOJO_PUBSUB_EXPERIMENTAL } = 1; Minion->new( mysql => MyJob::DBConnectionManager->new->get_connection_uri({ db_type => 'feeds', io_type => 'rw', })); }, ); 我们在 Minion 周围包含了一个简单的 Moose 类，以便使用我们想要的额外功能轻松添加到任何类或 Dancer 应用程序。

https://docs.databricks.com/_static/notebooks/complex-nested-structured.html 虽然这个深入的博客列出并解释了处理和处理复杂数据类型和格式的概念和动机，但这个笔记本示例通过几个具体示例检查了如何将它们应用于您在使用中可能遇到的数据类型案例。这个简短的笔记本教程展示了一些方法，您可以在其中探索和使用许多新的帮助程序 Spark SQL 实用程序函数和 API 作为org.apache.spark.sql.functions 包的一部分。特别是，它们在执行 Streaming ETL 时派上用场，其中数据是具有复杂和嵌套结构的 JSON 对象：嵌入为 JSON 的 Map 和 Structs： _ get_json_object() _ from_json() _ to_json() _ explode() _ selectExpr() 本简短教程的内容是使用 Spark SQL 实用程序函数对嵌套 JSON 结构进行切片和切块的无数方法。 让我们创建一个带有属性和值的简单 JSON 模式，而不使用任何嵌套结构。 import org.apache.spark.sql.types._ // include the Spark Types to define our schema import org.apache.spark.sql.functions._ // include the Spark helper functions val jsonSchema = new StructType() .add("battery_level", LongType) .add("c02_level", LongType) .add("cca3",StringType) .add("cn", StringType) .add("device_id", LongType) .

欢迎来到今年 12/25 的 Raku One-Liner Advent Calendar! 的第12天。今天，我们将研究一个计算零的单行程序。 say 0.1 + 0.2 - 0.3 如果您熟悉编程，那么您很清楚，只要开始使用浮点运算，就会失去精度，并且很快就会面对小错误。 您可能还看到了网站，0.30000000000000004.com，它有很多不同的编程语言列表，以及它们如何打印简单的表达式 0.1 + 0.2。在大多数情况下，您没有得到 0.3 这个准确值。通常当你得到零时，它实际上是在打印操作期间舍入的结果。 在 Raku 中，0.1 + 0.2正好是 0.3，今天的单行程序正好打印了一个零。 让我们稍微探讨一下 Raku 的内部结构，看看它是如何工作的。几天前，我们看到 Raku 的 grammar（在 Rakudo 编译器中实现）具有以下检测数字的片段： token numish { [ | 'NaN' >> | <integer> | <dec_number> | <rad_number> | <rat_number> | <complex_number> | 'Inf' >> | $<uinf>='∞' | <unum=:No+:Nl> ] } 假如你很熟悉 Raku，就会知道 Raku 使用有理数来存储浮点数（如0.1）这一事实可以解释上述行为。这是对的，但是看一下 grammar，你会发现这篇文章有点长。 Grammar 中所谓的 rat_number 是用尖括号写的数字：

现在我有一个 Minion 工作队列，我需要妥善处理它。工人是否在工作（他们是否抓住了生产资料）？工作是否顺利完成？有什么错误吗？这些是什么？ Minion Jobs Command Minion 带有一个 job 命令，列出了作业及其状态。 $ perl myapp.pl minion job 6 inactive default check_url 5 active default check_url 4 failed default check_url 3 failed default check_url 2 finished default check_url 1 finished default check_url 我可以通过传递工作的ID来查看单个工作的信息。 $ perl minion.pl minion job 1 { "args" => [ "http://mojolicious.org" ], "attempts" => 1, "children" => [], "created" => "2018-11-23T19:15:47Z", "delayed" => "2018-11-23T19:15:47Z", "finished" => "2018-11-23T19:15:48Z", "id" => 1, "notes" => {}, "parents" => [], "priority" => 0, "queue" => "default", "result" => "0.

package wmstat.trip import org.apache.hadoop.hbase.client._ import org.apache.hadoop.hbase.io.ImmutableBytesWritable import org.apache.hadoop.hbase.mapreduce.TableInputFormat import org.apache.hadoop.hbase.HBaseConfiguration import org.apache.hadoop.hbase.filter._ import org.apache.hadoop.hbase.protobuf.ProtobufUtil import org.apache.hadoop.hbase.util.{Base64, Bytes} import org.apache.spark.sql.SQLContext import org.apache.spark.{SparkConf, SparkContext} import wmutils.WmTimeUtil._ import scala.collection.mutable.ArrayBuffer object HBaseSpark { def main(args:Array[String]): Unit ={ // 本地模式运行,便于测试 val sparkConf = new SparkConf().setMaster("local").setAppName("HBaseTest") // 创建 HBase 扫描器 val scan = new Scan() // val filter=new RowFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL, new RegexStringComparator("^[a-zA-Z0-9]+_20180903[0-9]{6}$")) //使用正则表达式过滤近一个月的 // scan.setFilter(filter) // 过去 7 天 val arrayWeek: ArrayBuffer[String] = lastestNdays("", 7) // filterList val filterList = new FilterList(FilterList.

欢迎来到 Raku One-Liner Advent Calendar 的第11天！ 今天，日历文章完全致力于解决欧拉计划的问题34。 如果你想在看到我的答案之前找到自己的答案，请再次暂停阅读。 因此，任务是找到所有数字的总和，它们等于其数字的阶乘的总和。 听起来很清楚？ 🙂 你可以看一下单行程序的解决方案，以便更好地理解它。 say [+] (3..50_000).grep({$_ == [+] .comb.map({[*] 2..$_})}) 让我们从……开始吧。 我们循环范围 3 .. 50_000。上边界是基于某些要考虑因素的猜测。我不会在这里解释，但如果你好奇，你可能会试着找到答案。基本上，在某些时候你会理解这个数字要么包含太多数字，要么本身就太大了。请参阅有关 Project Euler 的讨论以获得纯理论的解释。 第二步是 grep。我们正在搜索与总和相等的数字（$_ ==）。它是通过第二个化简加号 [+] 计算出来的，但你可以使用 sum 方法代替： {$_ == .comb.map({[*] 2..$_}).sum} 请注意 .comb 是一个在默认变量 $_ 上调用的方法。 comb 方法将数字拆分为单独的数字（作为字符）。 map 方法将每个数字转换为阶乘（再次使用化简运算符 [*]，就像我们昨天所做的那样）。 最后，最外层的 [+] 将所有 grepped 的数字相加，并将结果传递给 say 例程。 虽然主要想法是展示一个单行，但在实际操作中，在使用它们之前准备因子是更明智的： my @f = (0..9).map({[*] 1..$_}); say [+] (3..50_000).grep({$_ == [+] .comb.map({@f[$_]})}); 这就是今天故事的结局。这个日历的其余部分还有更多内容！

Minion 作业队列是一个非常有用的工具，但有时我需要将它用于非 Web 项目。那么，如何在不需要 Mojolicious Web 应用程序的情况下使用 Minion？ 如果我没有足够大的工作需要 Beam::Minion 提供的任务组织，虽然 Minion 可以用作完全独立的库，但最简单的解决方案就是构建一个 Mojolicious 应用程序。幸运的是，一个 Mojolicious 应用程序可以只有2行： use Mojolicious::Lite; app->start; 现在，如果我从未运行过 web 守护进程，那么这绝不会是一个网站。那么，如果没有人使用网络浏览器访问它，这可以称为 Web 应用程序吗？🤔 Add Minion 要使用 Minion，请将 Minion 插件添加到应用程序。对于这个例子，我将使用 SQLite Minion 后端，这样我就不需要运行单独的数据库进程，但如果你有一个接受远程连接的数据库，Minion 可以在多台机器上工作。 plugin Minion => { SQLite => 'sqlite:' . app->home->child('minion.db'), }; 加载 Minion 插件后，我的应用程序获得了一些新功能： 我可以使用 minion.add_task 助手程序添加 Minion 任务（可运行的代码） 我可以通过多种方式排队工作： 从命令行使用 minion job 命令 从应用程序内部使用 minion.enqueue 助手程序 从任何 Perl 脚本加载 Minion 并使用 enqueue 方法 我可以使用 minion worker 命令运行一个 Minion worker，它将执行任何排队的作业 Create a Task 我将创建一个名为 check_url 的任务来检查下载 URL 所需的时间。 Time::HiRes 核心模块将为我提供高分辨率时间。

package wmstat.trip import org.apache.log4j.{Level, Logger} import org.apache.spark.sql.execution.datasources.hbase._ import org.apache.spark.sql.functions._ import org.apache.spark.sql.{DataFrame, SparkSession} import scopt.OptionParser import wmutils.WmTimeUtil._ import scala.collection.mutable.ArrayBuffer /** * 提交方式 * spark2-submit --name HBASE-CONNECTOR --files /etc/hbase/conf/hbase-site.xml --class wmhbase.MileageAnxiety --master yarn --deploy-mode client --driver-memory 2g --driver-cores 2 --executor-memory 2g --executor-cores 1 --num-executors 2 wmanxiety-1.0-SNAPSHOT.jar --day 20180810 --repartition 500 --interval 7 */ object MileageAnxiety { def cat = s"""{ |"table":{"namespace":"default", "name":"trip_signal", "tableCoder":"PrimitiveType"}, |"rowkey":"key", |"columns":{ |"rowkey" :{"cf":"rowkey", "col":"key", "type":"string"}, |"vin" :{"cf":"info", "col":"vin", "type":"string"}, |"tripStatus" :{"cf":"info", "col":"tripStatus", "type":"string"}, |"tripStartTime":{"cf":"info", "col":"tripStartTime", "type":"string"}, |"tripEndTime" :{"cf":"info", "col":"tripEndTime", "type":"string"}, |"tripDistance" :{"cf":"info", "col":"tripDistance", "type":"string"}, |"startSoc" :{"cf":"info", "col":"startSoc", "type":"string"}, |"endSoc" :{"cf":"info", "col":"endSoc", "type":"string"}, |"maxSpeed" :{"cf":"info", "col":"maxSpeed", "type":"string"}, |"startMileage" :{"cf":"info", "col":"startMileage", "type":"string"}, |"coordinate" :{"cf":"info", "col":"coordinate", "type":"string"} |} |}""".

欢迎来到 Raku One-Liner Advent Calendar 的第10天！今天，将有三个单行命令而不是一个。 我们今天的客人是一个带有方括号的化简结构。当它们不包围数组索引时，它们就工作在完全不同的领域中了。 例 1 最经典的例子，也是 Raku Calendar 2019，使用化简运算符来计算阶乘： say [*] 1..2019 Raku 中的 [ ] 是一个化简元运算符。名字里面的的「元」告诉我们它可以用作另一个运算符的包层（顺便说一句，不仅仅是作用于运算符）。 在第一个示例中，该运算符包含另一个运算符，并且可以通过将范围注册到列表并将 * 放在其所有元素之间来重写整行： say 1 * 2 * 3 #`(more elements) * 2018 * 2019 例2 现在，让我们解决欧拉项目的问题5，我们需要找到最小的数字，这个数字可以被从1到20的所有数字整除。 让我在 Raku 中给你一个直接的答案： say [lcm] 1..20 此代码看起来与前面的示例非常相似，但使用另一个运算符 lcm 例程，它是 Raku 中的中缀运算符。这个名字代表最小公倍数，但在文档中你还可以读到它返回这两个参数都可以被整除的最小整数。几乎相同的词，用于表达我们要解决的问题。 say 1 lcm 2 lcm 3 lcm 4 lcm 5 lcm 6 lcm 7 # ... and up to 20 例子3 其他已经内置在 Raku 中的中缀运算符也可以非常高效。这是一个只用几个代码字符旋转矩阵的例子：

你写了一个很棒的Mojolicious应用程序，人们使用它。奇妙！但是用户可能想要修改应用程序的主题：更改徽标，使用其他CSS框架，这类事情。 修改Mojolicious应用程序的主题非常简单：在公共和模板中添加，修改或删除内容。但是所有这些直接修改可能无法在更新应用程序时继续存在：它们只会被新版本的文件删除。 让我们看看我们如何提供一种在Mojolicious应用程序中拥有主题系统的方法，该方法允许用户在没有痛苦的情况下拥有自定义主题，并且没有丢失更新的风险。 一个新的应用程序 当您使用mojo生成MyApplication创建新的Mojolicious应用程序时，这些是将提供文件及其默认内容的默认目录： $ tree public templates public └── index.html templates ├── example │ └── welcome.html.ep └── layouts └── default.html.ep 2 directories, 3 files public是存储静态文件的地方，而templates是存储模板的地方。 这些路径在$ app-> static-> paths和$ app-> renderer-> paths中的Mojolicious应用程序中注册。幸运的是，这两个对象是数组引用，因此我们可以向它们添加或删除目录。 当提供静态文件时，我们的应用程序在$ app-> static-> paths数组的第一个目录中搜索该文件，如果找不到它，则搜索下一个目录，依此类推。模板渲染也是如此。 让我们改变路由 我们可以将公共模板和模板默认目录保存在应用程序目录的根目录下，但我喜欢在一个名为themes的目录中重新组合所有与主题相关的东西并调用我的默认主题…好吧，默认。 创建新目录并在其中移动默认主题目录： $ mkdir -p themes/default $ mv public templates themes/default 然后，我们需要更改应用程序中的路径。在 lib/MyApplication.pm 中添加： # Replace the default paths $self->renderer->paths([$self->home->rel_file('themes/default/templates')]); $self->static->paths([$self->home->rel_file('themes/default/public')]); 添加使用其他主题的方法 如前所述，Mojolicious在已注册路径的第一个目录中搜索静态文件或模板，如果找不到文件或模板则转到下一个。 因此，我们需要在默认主题路径之前添加新主题路径。 假设我们创建了一个圣诞主题，其中文件位于主题/圣诞节/公共内容中，哪些模板位于主题/圣诞节/模板中。 我们要添加到代码中的代码段变为： # Replace the default paths $self->renderer->paths([$self->home->rel_file('themes/default/templates')]); $self->static->paths([$self->home->rel_file('themes/default/public')]); # Put the new theme first unshift @{$self->renderer->paths}, $self->home->rel_file('themes/christmas/templates'); unshift @{$self->static->paths}, $self->home->rel_file('themes/christmas/public'); 通过这种方式，我们可以重载默认文件。

欢迎来到 Raku One-Liner Advent Calendar 的第9天！ 在第6天，我们有一个带有交叉算子的构造，(999...100) X* (999...100)。 今天，我们将从11月开始进入类似的结构： 1..10 X* 1..10 它打印乘积表的项，数字从1到10： (1 2 3 4 5 6 7 8 9 10 2 4 6 8 10 12 14 16 18 20 3 6 9 12 15 18 21 24 27 30 4 8 12 16 20 24 28 32 36 40 5 10 15 20 25 30 35 40 45 50 6 12 18 24 30 36 42 48 54 60 7 14 21 28 35 42 49 56 63 70 8 16 24 32 40 48 56 64 72 80 9 18 27 36 45 54 63 72 81 90 10 20 30 40 50 60 70 80 90 100) 关于这样的结构，有几点需要了解。

仍然有不少人在生产中使用LDAP，但对于那些不熟悉它的人来说，LDAP是一个具有树结构的目录，该树结构针对非常快速的查找进行了优化。它曾经是非常常见的集中式身份验证系统，如果您使用的是Active Directory，那么您（主要是）使用LDAP。我漫游在身份验证的荒野中，最后是我的解决方案，如何将LDAP身份验证添加到您的应用程序。 这篇文章是基于我在2018年伦敦Perl研讨会上发表的一篇演讲。我们有点乐观地认为他们会在圣诞节之前编辑所有视频，但我们希望能有一种顿悟。 LDAP只是验证周期的一小部分，所以这篇文章对于你必须编写自己的凭证检查器的情况来说相当好。哦，这篇文章的谈话和写作完全符合我的意图和提出的问题，我没有考虑过这些问题。结果，它开始像爱丽丝的餐厅一样，没有完整的编排和五部分和谐。我希望这个警示故事可以帮助你避免陷入同样的​​陷阱。 在此期间，请参加MojoConf 2018的Lightning Talk。 路由 - lib/MyApp.pm 首先，一个忏悔。我从未真正进入过Lite Apps。我知道很容易将它们发展为完整的应用程序，但是当我开始并且从未回到它时，我面临着制定解决方案的压力。结果是这篇文章是关于认证一个完整的应用程序，并不像其他帖子谈论他们的Lite应用程序那样苗条。 直接进入，我们假设您已经在模板中有一个登录页面，并且它有一个将数据发布到/ login的表单。如果你有这样的路线 $r->post('/login')->name('do_login')->to('Secure#on_user_login'); 将凭据发送到您的控制器。或者，如果你对命名路线很酷，你的模板可能会包含这一行 <form action="<%= url_for 'do_login' %>" method="POST"> 专业提示：你甚至可以简化它 %= form_for 'do_login' 如果路由只处理POST，它会为您完成所有操作，包括方法。 控制器 - lib / MyApp / Controller / Secure.pm 让我们从 Mojolicious Cookbook 中汲取灵感。 package MyApp::Controller::Secure; use Mojo::Base 'Mojolicious::Controller'; sub on_user_login { my $self = shift; my $username = $self->param('username'); my $password = $self->param('password'); if (check_credentials($username, $password)) { $self->render(text => 'Hello Bender!'); } else { $self->render( text => '<h2>Login failed</h2><a href="/login">Try again</a>', format => 'html', status => 401 ); } } sub check_credentials { my ($username, $password) = @_; return $username eq 'Bender' && $password eq 'rocks'; } 存储密码 - MojoX::Auth::Simple 我们同意硬编码用户名和密码是不可持续的。如果您可以连接到数据库，Perl DBI 模块可以连接的任何数据库，那么您可能会认为 MojoX::Auth::Simple 将解决您的问题。进一步阅读将告诉您它只提供帮助方法 log_in，is_logged_in 和 log_out，这些方法对于身份验证周围的所有内容都很有用，但对身份验证本身不起作用。但是，既然您现在正在使用数据库，那么您可以将 check_credentials 更改为比这更好的东西（wot是在周五下午制作而未经过测试）

欢迎来到今年的 Raku One-Liner Advent Calendar 的第8天。 它大约是整个系列的 ¼，并且不要忘记你可以在 Raku 中键入 ¼ 而不是0.25！ 今天，我们正在解决欧拉项目的问题2。 任务是找到所有偶数斐波纳契数都低于四百万的数字的总和。 这是完整的答案： (1, 1, * + * ... * > 4_000_000).grep(* %% 2).sum.say 从左侧或右侧解析代码同样有趣。 我们从左边开始吧。 在第一个圆括号内，我们生成一个斐波纳契数序列，从两个1开始，每个跟随的数字是前两个数之和。 在 Raku 中，您可以使用 WhateverCode 块表达它：* + * 相当于 {$^a + $^b}。 Raku 序列的一个鲜为人知的特征是最终条件。 在许多例子中，你会看到裸星或 Inf。 在我们的示例中，我们使用显式的上边界限制序列。 请注意，您不能简单地写成这样： 1, 1, * + * ... 4_000_000 要更好地将其可视化，请尝试更小的限制，例如100： > (1, 1, * + * ... 100) (1 1 2 3 5 8 13 21 34 55 89 144 233 377 610 987 1597 2584 4181 当序列越过我们想要的边界时，Raku 没有停止，并继续生成数字。 只有当序列的下一个计算元素完全等于给定数字时，它才会停止它，例如：

在这几年meta :: hack 3中，我非常幸运地与Joel Berger合作，通过Mojolicious将OpenAPI与MetaCPAN API集成/记录。 它是什么？ OpenAPI是用于设计，记录，验证和驱动RESTful API的规范。它可用于为现有API提供文档，或者在创建新API时提供文档。 OpenAPI规范起源于Swagger规范，并重命名为将API描述格式（OpenAPI）与开源工具（Swagger）分开。该规范已移至新的GitHub存储库，但未更改。 对于MetaCPAN API，我们开始提供现有API的文档，但很快也转向支持API调用的验证。 工具 OpenAPI有许多可用的工具可供帮助，包括有助于编写规范的发现工具。我们选择手动编写定义（当然是vim）并使用工具生成文档并将规范集成到MetaCPAN中。 ReDoc - OpenAPI / Swagger生成的API参考文档 ReDoc创建一个交互式页面，根据OpenAPI规范文件中提供的详细信息提供文档和示例。 ReDoc包含一个HTML模板，用作静态文件，用于自定义文档的显示方式。 Mojolicious :: Plugin :: OpenAPI - Mojolicious的OpenAPI / Swagger插件 读取OpenAPI规范文件，并为基于Mojolicious的应用程序添加适当的路由和验证。 JSON :: Validator - 根据JSON模式验证数据 集成到Mojolicious :: Plugin :: OpenAPI模块中，提供输入和输出验证，以及为规范文件本身提供验证。 入门 在实现MetaCPAN OpenAPI规范时使用以下策略。 OpenAPI 规范文件 通过支持多行属性值，可以更轻松地以更少的格式进行读写，我们选择了YAML。还支持JSON。 # Define the version of the OpenAPI spec to use. Version 2.0 still uses # swagger as the key swagger: "2.0" # general information about the API info: version: "1.

欢迎来到 Raku One-Liner Advent Calendar 的第7天！ 今天，我们将看看2019年Raku日历中的三月份： 这里的代码使用 ASCII 领地之外的三个字符。 我们甚至可以再添加一个 Unicode 字符： say π × $𝜌² 在 Raku 中，您可以在标识符中自由地使用 Unicode 字符，例如变量或函数名。 但最重要的是，有许多预定义的符号，如 π，它们有 ASCII 替代品。 检查文档页面 Unicode 与 ASCII 符号，以查看可在 Raku 中使用的全部 Unicode 字符集。 使用 ASCII，可以通过以下方式重写上述单行程序： say pi * $r ** 2 让我们回到第2天的附加题代码，看看 Unicode 字符可以用在什么位置： sub f($n) { ($n <<*>> (1...1000 / $n)).grep: * < 1000 } say (f(3) ∪ f(5)).keys.sum; 这里有一些因素。 首先，超运算符 <<*>> 可以用适当的法语引号替换：«*»，乘法字符可以是我们今天使用的交叉：«×»。 同样可以应用于除法：÷。 其次，序列运算符的三个点可以用一个 Unicode 字符替换：…（如果您在 Word 中编程，则在键入完整停止三次后自动获得此字符）。

在这些近代，世界上有数十亿人，圣诞老人需要一个现代化的系统来跟踪他的淘气又好看的名单。幸运的圣诞老人，现代 Perl 有一个现代的 web 框架，Mojolicious。 第1步：构建列表 首先，我们需要一个数据库模式。圣诞老人只需要知道某人是否顽皮或好，所以我们的架构非常简单。我们将通过使用 Mojo::SQLite 连接到 SQLite 数据库并使用 Mojo::SQLite 迁移从我们脚本的DATA部分加载我们的数据库模式来启动我们的 Mojolicious::Lite 应用程序： use v5.28; use Mojolicious::Lite; use Mojo::SQLite; # Connect to the SQLite database and load our schema from the # '@@ migrations' section, below my $db = Mojo::SQLite->new( 'sqlite:thelist.db' ); $db->auto_migrate(1)->migrations->from_data(); # Start the app. Must be the last code of the script. app->start; __DATA__ @@ migrations -- 1 up CREATE TABLE the_list ( id INTEGER PRIMARY KEY AUTOINCREMENT, name VARCHAR NOT NULL, address VARCHAR NOT NULL, is_nice BOOLEAN DEFAULT FALSE, is_delivered BOOLEAN DEFAULT FALSE ); 创建我们的架构后，我们可以添加 Yancy。 Yancy是一个简单的 CMS，用于编辑内容和简化数据驱动的网站开发。我们将告诉 Yancy 阅读我们的数据库架构以进行自我配置，但我们也会给它一些提示，以便更轻松地编辑内容。

欢迎来到 Raku One-Liner Advent Calendar 的第6天！ 正如昨天所承诺的那样，今天我们将解决欧拉项目的问题4。 让我再次提醒您，您可以先自己暂停阅读并自行解决问题。 我的目的是展示Raku和Perl的美妙之处。 因此，任务是找到最大的回文数（从两端读取的数字，例如1551），这是两个三位数的乘积。 换句话说，我们必须扫描999×999以下的数字，并且可以优化解决方案，但实际上，我们只需要允许数字，即乘积，因此，我们不要跳过乘法部分。 这是我们今天的单行程序： (((999...100) X* (999...100)).grep: {$^a eq $^a.flip}).max.say 如果你正在阅读这个 advent 日历，那么你已经准备好了链式方法调用在Raku单行中使用非常方便的事实。 我们在第2天早些时候也看到了 grep 的冒号形式，但这次我们使用了一个带占位符变量的代码块。 目前还不清楚你是否可以在这里使用星号，因为我们需要在块中使用两次变量。 该行的第一部分使用十字运算符 X*，我们将在几天内回到本系列中。 它生成所有三位数字的乘积。 由于我们需要最大的数字，从右到左开始是有意义的，这就是为什么序列 999 ... 100，而不是 100 ... 999。 让我们看看 grepped 乘积序列中的前几个数字： 580085 514415 906609 119911 282282 141141 853358 650056 单行并不总是非常理想。 在我们的例子中，我们需要生成整个乘积序列以找到它们中的最大值。 答案位于第三个位置，因此将 max 替换为 first 将是错误的。 但好的部分是，如果你先使用，Raku将不会生成所有数字。 还有另一个有用的方法，head，它也可以防止生成超过必要结果的方法。 以下代码运行得更快，并提供正确的结果： (((999...100) X* (999...100)).grep: {$^a eq $^a.flip}).head(10).max.say 让我们今天就到此为止。 敬请关注！

当人们告诉我我不能（他们的意思是不应该）用正则表达式解析 HTML 时，我说“拿酒来”。这不仅仅是技巧或态度，而是方便。以正确的方式做到并不总是那么容易（我记得 HTML 0.9 是一个大问题）。最近，我一直在使用 Mojo::DOM 为我做这件事。它比旧的，权宜之计更容易。 诀窍始终是隔离感兴趣的 HTML。我可以做到这一点切除有趣部分周围的所有数据： my $html = ...; $html =~ s/.*?<table class="foo".*?>//; $html =~ s/<\/table>.*//; 现在我不必解析所有的 HTML;我可以考虑一下表格。即使这是权宜之计，也不是那么好。在我用更好的东西替换它之前，我会快速绕道而行。 层叠样式表 您可能知道层叠样式表（CSS）使你的网页看起来很漂亮（但不是我的，真的）。您可以将元数据添加到标记上： <img id="bender" class="robot" src="..." /> <img id="fry" class="human" src="..." /> <img id="leela" class="mutant" src="..." /> CSS 规则可以通过其 ID 或类来处理这些项目以将样式应用于它们。这个寻址是一个“选择器”，拥有比我更好的技能的人使用这些来使演示非常漂亮： img#fry { border: 1px; } img.robot { margin: 20px; } HTML 可能有点复杂。也许那些有趣的标签在列表中。这包裹了数据的另一层 HTML 结构： <ul class="employees"> <li><img id="bender" class="robot" src="..." /></li> <li><img id="fry" class="human" src="..." /></li> <li><img id="leela" class="mutant" src=".

欢迎来到这个可爱的 Raku One-Liner Advent Calendar 的第5天！ 今天，我们将回答今天是什么日期的问题（如果你愿意的话，明天我们可以谈论回文）。 因此，要打印答案，您可以使用以下Raku代码行： DateTime.now.yyyy-mm-dd.say 它看起来显而易见，并以YYYY-MM-DD的格式打印日期。 好的部分是 DateTime 类可以直接使用，例如，您不需要像在Perl 5中那样包含模块。 $ raku -e'DateTime.now.yyyy-mm-dd.say' 2018-12-05 正如您在日历的前几天已经看到的那样，链式方法调用是典型的Raku惯用法。 另一种选择是使用say作为子例程并使用括号来标记方法调用： say(DateTime.now().yyyy-mm-dd()); 这段代码也有效; 它完全正确，但看起来很重。 您还应该注意并告诉您的朋友，在Raku中，您可以在标识符中使用短划线和撇号。 好吧，也许使用撇号不是一个好主意，但是在Raku的源代码中已经广泛使用了连字符。只需确保在表达式中的减法运算符周围放置空格以避免解析中的任何冲突。 浏览源代码或阅读文档，您将找到以相同方式命名的另一种方法： hh-mm-ss。 我打赌你明白它的作用。 > DateTime.now.hh-mm-ss.say 00:12:01 请注意，对于不同的输出格式，您将找不到类似的方法，例如 dd-mm-yy 或 hh-mm。 请改用 formatter 程序。 它不是方法，而是在 Datish 角色中定义的属性。 DateTime 类中有一个默认格式化程序，但您可以通过为构造函数提供自己的子例程来重新定义它，例如： DateTime.now(formatter => -> $dt { sprintf '%02d.%02d.%04d', $dt.day, $dt.month, $dt.year }).say 这里的格式化程序接收一个带有参数 $dt 的匿名子例程（由箭头引入）。 我希望这段代码与我们的初始单行打印相同的日期，因为您最有可能在一天内阅读整篇文章。 不过，明天见一下测试回文的代码！

Test::Mojo 是 Mojolicious 测试工具，有很多方法可以测试 Web 应用程序中的路由（甚至可以在其他 Web 框架中进行测试）。 但是，如果我需要测试的不是路由呢？如果它是一个钩子，一个插件或一个助手怎么办？我们也可以测试所有这些东西！ 钩子 为了彻底测试钩子，我需要找到配置我的测试用例的方法。我可以指望我的应用程序执行它，并找到正确的路由来测试正确的行为。但是，这会产生更大的测试，集成不同的部分并使测试失败更难调试。我想要的是隔离我正在测试的东西。最好的方法是创建仅测试我想测试的路由。 如果我有一个钩子来记录特殊日志文件的异常，如下所示： #!/usr/bin/env perl use Mojolicious::Lite; # Log exceptions to a separate log file hook after_dispatch => sub { my ( $c ) = @_; return unless my $e = $c->stash( 'exception' ); state $path = $c->app->home->child("exception.log"); state $log = Mojo::Log->new( path => $path ); $log->error( $e ); }; app->start; 为了测试这个，一旦我加载了我的应用并创建了一个 Test::Mojo 对象，我就可以自由地为我的应用添加更多配置，包括新路由！这些路由可以为我的测试设置恰当的条件。 # test.pl use Test::More; use Test::Mojo; use Mojo::File qw( path ); my $t = Test::Mojo->new( path( 'myapp.

欢迎来到第4天的 Raku One-Liner Advent Calendar! 今天我们会看一看欧拉项目的第十三个问题。我先给你展示一张截图： 实际上，它看起来很大，任务是找到一百个整数之和的前十位数，每个整数由50位数组成。 听起来像是一项任务，可能需要一些优化和简化来摆脱对结果的前十位无贡献的一切。 但不是在Raku中。 在Raku中，您可以简单地将数字相加并取其前十位数字： < 37107287433902102798797998220837590246510135740250 # Other 98 numbers here 53503534526472524250874054075591789781264330331690 >.sum.substr(0, 10).say Raku默认使用任意长整数运行; 您不需要包含任何模块或以其他方式激活此行为。 您甚至可以快速计算幂并获得结果： $ raku -e'say 37107287433902102798797998220837590 ** 1000' 需要注意的另一件事是我们可以透明地将整数转换为数字，反之亦然。 在我们今天的程序中，数字列表显示为一对尖括号内的字符串的引用列表。 在列表中，您调用 sum 方法，该方法适用于数字。 获得总和后，再次将其视为字符串并提取其前十个字符。 整个代码看起来非常自然，易于阅读。 有了这种情绪，我们明天再见！

通过组合 Promises 来创建新的复杂 Promise Mojolicious 7.49 添加了自己的 Promises/A+ 规范实现。莫霍克在第14天写到了这些：你答应召唤！ 在 2017 年的 Mojolicious Advent Calender 中，他向你展示了如何同时获取多个网页。这个 Advent 条目扩展了更多的 Promise 技巧。 Promise 是一种旨在消除嵌套回调（也称为“回调地狱”）的结构。正确编写的 Promises 链具有扁平结构，易于线性跟随。 高阶 Promise 是包含其他 Promise 并以其身份为基础的 Promise。 Mojo::Promise::Role::HigherOrder 发行版提供了三个角色，您可以将它们混合到 Mojo::Promise 中以创建包含 Promises 的更好看的 Promise。不过，在你看到它们之前，先看看 Mojo::Promise 已经提供的两个。 All 只有当 all Promise 都解决时，所有承诺才会解决。如果其中一个被拒绝，则拒绝所有承诺。这意味着如果一个被拒绝并且不需要知道任何其他人的状态，整个 Promise 知道该怎么做。 use Mojo::Promise; use Mojo::UserAgent; my $ua = Mojo::UserAgent->new; my @urls = ( ... ); my @all_sites = map { $ua->get_p( $_ ) } @urls; my $all_promise = Mojo::Promise ->all( @all_sites ) ->then( sub { say "They all worked!

欢迎来到 Raku One-Liner Advent Calendar 的第3天！ 今天，我们将生成随机数。 您可能会问，与它有什么关系，调用一种 rand 函数不是常规任务吗？ 好吧，从某种意义上说，是的，但在Raku中，您可能更喜欢调用方法。 我们来看看Raku日历中8月的代码： 2019.rand.Int.say 这就是整个程序！ 它只使用方法调用，方法一个接一个地链接在一起。 如果您以前从未见过Raku，那么你首先注意到的是在一个数字上调用的方法。 在Raku中，这一点也不特别。 在数字上调用 rand 方法可能是对 Cool类 中定义的方法的调用，该类立即代理给数字的数字表示： method rand() { self.Num.rand } 稍后在 Num类 中，会发生对底层 NQP 引擎的调用： method rand(Num:D: ) { nqp::p6box_n(nqp::rand_n(nqp::unbox_n(self))); } 在我们的示例中，对象 2019 是 Int，因此 rand 被直接分派给 Num 类的方法。 rand 方法返回一个浮点数，因此在其上调用 Int 来获取整数。 运行代码几次以确认它生成随机数： $ raku -e'2019.rand.Int.say' 543 $ raku -e'2019.rand.Int.say' 1366 $ raku -e'2019.rand.Int.say' 1870 如果您想了解随机数生成器的质量，请深入挖掘 NQP 和 MoarVM，然后再深入了解虚拟机的后端引擎。 要使结果可重复（例如，对于测试），通过调用 srand 函数来设置种子：

使用 Mojolicious 开发 Web 应用程序非常有趣！使用 morbo 服务器进行开发，对我的 webapp 的每次更改都会导致重新启动以加载我的更改。这样我下一个请求就有了我的新代码！ 所以，我改变了我的代码，重新启动了 webapp，然后我回到浏览器窗口。等等……我的新代码在哪里？为什么不修复错误？没有…我忘记重新加载我的浏览器窗口了吗？啊!当然！ 这是否发生在你身上？可能不是。但是，在每次后端代码更改后重新加载浏览器窗口仍然很烦人。如果每次重新启动 Web 服务器时我的浏览器窗口都会自动重新加载，那就太好了！ AutoReload 插件 像 Perl 中的每个问题一样，有一个 CPAN 模块：Mojolicious::Plugin::AutoReload。将此插件添加到我们的应用程序将自动重新加载连接到我们的应用程序的任何浏览器窗口，使开发 Mojolicious 应用程序更容易！ 要使用该插件，我们使用 plugin 方法将其添加到我们的应用程序中。然后，我们将 auto_reload 帮助器添加到布局模板中: use Mojolicious::Lite; plugin 'AutoReload'; get '/' => 'index'; app->start; __DATA__ @@ layouts/default.html.ep %= auto_reload %= content @@ index.html.ep % layout 'default'; <h1>Hello, World!</h1> 在这里下载代码。现在，当我们在浏览器中打开应用程序时，如果服务器重新启动，我们的浏览器将重新加载页面以查看新应用程序！ 工作原理 这个插件简单优雅：当浏览器加载页面时，它连接到位于 /auto_reload 的 WebSocket。服务器重新启动时，WebSocket 连接中断。客户端看到断开的连接，等待服务器再次开始侦听连接，然后重新加载页面。 在生产中禁用 一旦我们从 morbo 切换到 hypnotoad，我们就不再需要自动重新加载了。因此，插件不会向浏览器发送 JavaScript 来构建 websocket。这是使用 mode 属性控制的。当 mode 为 development 模式（morbo 的默认模式）时，将告知浏览器创建 WebSocket。当 mode 为其他任何模式时，不会创建 WebSocket。

欢迎来到 Raku One-Liner Advent Calendar 的第2天！ 今天, 我们将从编号为 1 的Euler 项目开始解决一个很好的任务. 我要再次提醒你, 该文的其余部分包含一个答案, 所以欢迎你暂停一下, 先考虑自己的答案。 但我几乎可以肯定, 如果你一直在阅读 raku.online, 那么你很可能在过去就已经解决了欧拉项目的问题。 任务是找到 1000 以下, 既是3的倍数, 又是5的倍数的数字的总和。我们感兴趣的行的前几个元素是 3,5,9,15,20,21等。你已经可以看到它们中的一些数字, 例如 15, 是 3和5的倍数, 因此不可能将3和5的倍数分别相加。 答案在这里: say sum((1..999).grep: * %% (3 | 5)) 现在让我们解读它。 我们需要的是过滤3的倍数或5的倍数的数字。如果你重新阅读上一句话, 铃声应该为你响起:在 Raku 中, 这可以用 Junctions 来实现, 非正式地称为量子叠加。 要测试数字是否可以被3或5整除, 请这样写: $x %% 3 | 5 顺便说一下, %% 是 divisibility 运算符, 是一个非常好的东西, 有助于避免布尔测试中的否定, 如果你写成下面这样, 那么你只有一个百分号: !($x % (3 | 5)) 好的, 主要条件已准备就绪, 让我们扫描1和（包括）999之间的数字:

欢迎来到 Mojolicious Advent Calendar 的另一年！ 2018 年对 Mojolicious 非常好，我认为没有比在挪威奥斯陆举行的 2018 年北欧 Perl 研讨会和 MojoConf 的回顾更好的方式开始这个日历了。 到挪威西部的边路 就个人而言，我特别喜欢这次旅行，因为我可以多花几天时间游览挪威西部，那里是标志性的峡湾。在乘船和火车游览峡湾之前，我拜访了卑尔根及其几个 Perlers。如果我的行李还没决定要在冰岛停留额外的几天，我会有一个延迟的停留，那将是一次完美的旅程。 我仍然度过了一段美好的时光，一生只见过一次！感谢克里斯托弗和乔尼斯，所有人都是我旅行的同伴！ Nordic Perl Workshop和 MojoConf Oslo Perl Mongers 知道如何抛出 Perl 事件！今年我实际上曾经两次参加 Perl 工具链峰会。 顾名思义，Nordic Perl Workshop 是 Perl 热情聚集和谈论 Perl 的区域研讨会。然而今年，它与 MojoConf 2018 联合品牌;之前的 MojoConf 也于 2014 年在奥斯陆举行。 在 Mojolicious 和 Perl 上都有很多很棒的演讲。您可以在 NPW Youtube 频道上看到它们。为简洁起见，我将在这里强调几点。 Keynote 今年的主题演讲由作家和 Perl 杰出人物 brian d foy 发表。他谈到了 Mojo 在他的咨询演出中如何迅速完成真正的工作。 Mojolicious 8.0 公告 Mojolicious 项目创始人兼首席开发商 Sebastian Riedel 颁发了“八点哦”。虽然它表面上是 8.

欢迎来到全新的 Raku.Online Advent Calendar 的第1天！今年，它的主题是 Raku One-Liners（来自 raku.onliners 的双关语）。因此，欢迎参加今年的 Raku One-Liner Advent Calendar。整个 raku.online 博客最初计划每天一篇，因此这是一个很好的机会，可以保持 25天每天一篇。 不要忘记关注其他人编写的其他与 Perl 相关的日历： Perl Advent Calendar Raku Advent Calendar Dancer Advent Calendar Mojolicious Advent Calendar 在接下来的几天里，我将发表简短的文章，要么解释 Raku Calendar 2019 的代码，要么解决 Project Euler 启发的一些问题，Raku 可以展示它的极致美感（如果你不想看到我的答案）在你之前，我邀请你先自己解决这些任务，然后再阅读这些文章。但是今天，让我们从其他东西开始，在 Raku 中生成一个随机密码。 这是完整的代码： ('0'..'z').pick(15).join.say 运行几次： Z4x72B8wkWHo0QD J;V?=CE84jIS^r9 2;m6>kdHRS04XEL O6wK=umZ]DqyHT5 3SP\tNkX5Zh@1C4 PX6QC?KdWYyzNOc bq1EfBZNdK9vHxz 每次运行时，我们的这行代码都会生成不同的字符串。如果您尚未安装 Raku，请自行尝试创建密码或使用上述密码之一。 前段时间，我们研究了pick方法的内部结构。它在Any类中定义，它实际上是将工作代理给List类的方法： proto method pick(|) is nodal {*} multi method pick() { self.list.pick } multi method pick($n) { self.

在我的上一篇文章中，我解释了我为生活所做的事情，并举例说明了我在工作中使用 Raku 的特殊内容。 这是另一个例子。 首先是一些背景信息：挪威是一个小国家，大多数人都讲同一种语言（当然，方言有所不同，但语言是相同的）。但听起来很疯狂……这种单一的口语有两种书面语言。是的，你读的是正确的：你说挪威语，但你写的是 Bokmål 或 Nynorsk - 松散地翻译成 “Book language” 和 “New Norwegian”。我不会深入了解其中的历史原因 - 如果你有兴趣我会推荐关于它的维基百科文章。我只想说这两个版本的挪威语非常相似，但差异足以让差异显而易见。 一些新闻媒体使用 Nynorsk，而大多数人使用 Bokmål。这样做的结果是，Bokmål 读者的内容远远多于 Nynorsk 读者。这就是为什么我的工作场所试图获得将 Bokmål 翻译成 Nynorsk 的软件的原因。 翻译工作得很好，但偶尔会有不幸。其中相当一部分是由译者误解的名字引起的。我们有字典的字典，所以它们应该很容易识别。但是，如果名称是较少使用的名称，则不容易识别。当字典失败时，我们可以尝试使用规则/正则表达式来识别它们。但即便如此，我们也无法总是决定一个单词是常规单词还是名称。 让我们来看看 Fuglesang 这个名字（即宇航员）。 Fuglesang 是一个名字，但它也是一个名词，意思是鸟的歌或唧唧。如上所述，如果使用正则表达式或使用名称字典在句子中间发生 Fuglesang 作为名称，则没有问题。正则表达式查找以大写字母开头的单词，并向译者发出该单词可能是名称的信号。但如果这个词出现在一个句子的开头 - 所有单词都被大写 - 那么 Fuglesang 就会变得模棱两可，即使这个名字出现在名字词典中。消除这种歧义很重要，因为如果没有，名称可以翻译为好像是任何其他单词。在我的情况下，由于不公正，Fuglesang 可能被翻译成 “fuglesong”，如果它是任何其他类型的单词而不是名称，这是正确的。 那么 Fuglesang 是什么？一首名字还是鸟鸣？ 显然，这是翻译软件需要帮助的情况。我想了一会儿，并决定也许我们可以开发一个相当天真的算法来帮助我们。我编写了一个简单的脚本：它解析一个文本并找到句子中出现的每个大写单词。然后它提取任何句子的第一个单词并检查该单词是否出现在句子中间单词列表中。如果确实如此，那么我们假设 - 在该单个文本的特定上下文中 - 该单词可以被视为名称。 我必须照顾的唯一特殊情况，与所有格有关。在英语中，所有格相对容易处理，因为它们很容易被发现和移除。假设你有两句 “Peter owns a car” 和 “This is Peter’s car”。一个简单的 Str.subst("'s", "") 将删除后一句中的所有格，现在 Peter 可以在第一句中与 Peter 相比较。 但是在挪威，所有格只是在没有撇号的情况下拖尾s-es，因此答案并不那么简单（“Peter eier en bil” vs “Peters bil”）。但总而言之，使用 Raku 解决这个问题也相当简单。

在本系列结构化流博客文章的第1部分中，我们演示了使用结构化流编写端到端流式 ETL 管道是多么容易，它将 JSON CloudTrail 日志转换为 Parquet 表。 该博客强调，构建此类管道的主要挑战之一是从各种来源和复杂格式中读取和转换数据。 在这篇博文中，我们将更详细地研究这个问题，并展示如何使用 Apache Spark SQL 的内置函数来解决所有数据转换挑战。 具体来说，我们将讨论以下内容： 有哪些不同的数据格式及其权衡 如何使用Spark SQL轻松使用它们 如何为您的用例选择正确的最终格式 Data sources and formats 数据以各种不同的格式提供。 电子表格可以用 XML，CSV，TSV 表示; 应用程序指标可以用原始文本或 JSON 写出。 每个用例都有针对它定制的特定数据格式。 在大数据领域，我们通常会遇到 Parquet，ORC，Avro，JSON，CSV，SQL 和 NoSQL 数据源以及纯文本文件等格式。 我们可以将这些数据格式大致分为三类：结构化，半结构化和非结构化数据。 让我们试着了解每个类别的好处和缺点。 结构化数据 结构化数据源定义数据的模式。利用有关底层数据的额外信息，结构化数据源可提供有效的存储和性能。例如，Parquet和ORC等柱状格式使得从列的子集中提取值变得更加容易。首先逐行读取每个记录，然后从感兴趣的特定列中提取值可以读取比查询仅对一小部分列感兴趣时所需的更多数据。基于行的存储格式（如Avro）可有效地序列化和存储提供存储优势的数据。然而，这些优点通常以灵活性为代价。例如，由于结构的刚性，演变模式可能具有挑战性。 非结构化数据 相比之下，非结构化数据源通常是自由格式文本或二进制对象，其不包含标记或元数据（例如，CSV文件中的逗号），以定义数据的组织。报纸文章，医疗记录，图像blob，应用程序日志通常被视为非结构化数据。这些类型的源通常要求数据周围的上下文是可解析的。也就是说，您需要知道该文件是图像或是报纸文章。大多数数据源都是非结构化的。具有非结构化格式的成本是从这些数据源中提取价值变得麻烦，因为需要许多转换和特征提取技术来解释这些数据集。 半结构化数据 半结构化数据源是按记录构建的，但不一定具有跨越所有记录的明确定义的全局模式。结果，每个数据记录都使用其架构信息进行扩充。 JSON和XML是很受欢迎的例子。半结构化数据格式的好处是，它们在表达数据时提供了最大的灵活性，因为每条记录都是自我描述的。这些格式在许多应用程序中非常常见，因为存在许多用于处理这些记录的轻量级解析器，并且它们还具有人类可读的优点。但是，这些格式的主要缺点是它们会产生额外的解析开销，并且不是特别为ad-hoc查询而构建的。 与Spark SQL交换数据格式 在我们之前的博客文章中，我们讨论了如何将Cloudtrail Logs从JSON转换为Parquet，将我们的即席查询的运行时间缩短了10倍。 Spark SQL允许用户从批处理和流式查询中提取这些数据源类中的数据。 它本身支持以Parquet，ORC，JSON，CSV和文本格式读取和写入数据，并且Spark包中还存在大量其他连接器。 您还可以使用JDBC DataSource连接到SQL数据库。 Apache Spark可用于交换数据格式，如下所示： events = spark.readStream \ .format("json") \ # or parquet, kafka, orc... .option() \ # format specific options .

Cassandra Sink for Spark Structured Streaming 我最近开始使用 Spark，并且必须将结构化流式 API 生成的结果存储在 Cassandra 数据库中。 在这篇文章中，我提供了一个如何在 Spark Structured Streaming 中创建和使用 Cassandra 接收器的简单示例。 我希望它对那些刚刚开始使用 Structured Streaming API 并想知道如何将它与数据库连接的人有用。 应用程序的想法非常简单。 它从 Kafka 读取消息，解析它们并将它们保存到 Cassandra 中。 Why Structured Streaming? 从文档中可以看出，Structured Streaming 是一个基于 Spark SQL 引擎的可扩展且容错的流处理引擎。 您可以使用 Dataset/DataFrame API 来表示流聚合，事件时间窗口，流到批处理连接等。在 Spark 中进行流式计算，并允许使用熟悉的 SQL 处理流数据。 What Is the Problem? Spark Structured Streaming 在 2017 年已经被标记为稳定。因此，它是一个相对较新的 API，并不是所有功能都在那里。例如，有几种类型的内置输出接收器：File，Kafka，Console 和内存接收器。但是，如果您想将流式计算的结果输出到数据库中，则需要使用 foreach 接收器并实现 ForeachWriter 接口。从 Spark 2.3.1 开始，这仅适用于 Scala 和 Java。 在这里，我假设您已经大致了解 Structured Streaming 如何工作，并且知道如何读取和处理流数据，现在可以将其输出到数据库中。如果上述某些步骤不清楚，可以使用一些很好的在线资源来帮助您开始使用结构化流。特别是，官方文档是一个很好的起点。在本文中，我想关注您需要将结果存储在数据库中的最后一步。

http://vishnuviswanath.com/spark_structured_streaming.html Structured Streaming 是 Apache Spark 的流式引擎, 可用于进行近实时分析。 在这篇博客中，我们通过一个非常简单的用例来探索结构化流。 想象一下，你开始一段旅程，需要检查车辆是否超速。 我们将创建一个简单的近实时流应用程序来计算每几秒钟车辆的平均速度，同时谈论 SlidingWindow，TumblingWindow，EventTime，ProcessingTime，Watermarks 和 Kafka Source＆Sink。 此博客中使用的所有代码都可以在我的 Github仓库中找到。 基于流的微批处理 在我们进入用例之前，让我们先看看 Apache Spark 中流式传输的工作原理。 Spark 中的结构化流，与其前身（DStream）类似，使用微批处理进行流式处理。 也就是说，spark 等待非常小的间隔，例如 1 秒（或甚至 0 秒 - 即，尽快），并将在该间隔期间接收的所有事件批量聚集成微批。 然后，Driver 程序将调度此微批处理在 Executors 中作为 Tasks 执行。 完成微批处理后，将收集下一批并再次调度。 频繁进行该调度以给出流式执行的印象。 Kafka Source 我们将读取 Kafka topic - cars 中的事件。 为此，我们需要将 format 设置为 “kafka”，用 boker 地址设置 kafka.bootstrap.server 并使用 “subscribe” 选项提供 topic 名。 val df: DataFrame = spark .readStream .format("kafka") .option("kafka.bootstrap.servers", "localhost:9092") .option("subscribe", "cars") //.

读取 parquet 文件发到 Kafka 定义 schema 根据 val df = spark.read.parquet("xxxx/part-*") df.printSchema 得到的 schema 如下： root |-- alm_dm_list: array (nullable = true) | |-- element: integer (containsNull = true) |-- alm_esd_list: array (nullable = true) | |-- element: integer (containsNull = true) |-- alm_others_list: array (nullable = true) | |-- element: integer (containsNull = true) |-- data_batt_sc_volt_lowest: double (nullable = true) |-- veh_dcdcst: integer (nullable = true) |-- alm_common_dcdc_st: integer (nullable = true) |-- dm_cnt: integer (nullable = true) |-- alm_common_temp_diff: integer (nullable = true) |-- data_batt_temp_probe_highest_seq: integer (nullable = true) |-- alm_common_dm_temp: integer (nullable = true) |-- alm_common_soc_high: integer (nullable = true) |-- alm_eng_cnt: integer (nullable = true) |-- alm_common_esd_high: integer (nullable = true) |-- alm_common_esd_unmatch: integer (nullable = true) |-- alm_common_soc_low: integer (nullable = true) |-- alm_esd_cnt: integer (nullable = true) |-- veh_curr: double (nullable = true) |-- esd_temp_data: array (nullable = true) | |-- element: struct (containsNull = true) | | |-- esd_temp_probe_cnt: integer (nullable = true) | | |-- esd_temp_probe_list: array (nullable = true) | | | |-- element: integer (containsNull = true) | | |-- esd_temp_subsys_seq: integer (nullable = true) |-- esd_temp_subsys_cnt: integer (nullable = true) |-- veh_pedal_deep: integer (nullable = true) |-- esd_volt_data: array (nullable = true) | |-- element: struct (containsNull = true) | | |-- esd_curr: double (nullable = true) | | |-- esd_frame_sc_cnt: integer (nullable = true) | | |-- esd_frame_sc_list: array (nullable = true) | | | |-- element: double (containsNull = true) | | |-- esd_frame_start: integer (nullable = true) | | |-- esd_sc_cnt: integer (nullable = true) | | |-- esd_volt: double (nullable = true) | | |-- esd_volt_subsys_seq: integer (nullable = true) |-- alm_others_cnt: integer (nullable = true) |-- data_batt_sc_volt_highest: double (nullable = true) |-- veh_chargest: integer (nullable = true) |-- alm_common_esd_charge_over: integer (nullable = true) |-- ts: long (nullable = true) |-- alm_common_sc_consistency: integer (nullable = true) |-- eng_consumption: double (nullable = true) |-- veh_pedalst: integer (nullable = true) |-- data_batt_subsys_temp_lowest_seq: integer (nullable = true) |-- loc_lon84: double (nullable = true) |-- esd_volt_subsys_cnt: integer (nullable = true) |-- data_batt_temp_lowestest: integer (nullable = true) |-- data_batt_subsys_volt_lowest_seq: integer (nullable = true) |-- alm_common_brk: integer (nullable = true) |-- alm_common_esd_low: integer (nullable = true) |-- veh_insulation: integer (nullable = true) |-- veh_volt: double (nullable = true) |-- alm_common_temp_high: integer (nullable = true) |-- alm_dm_cnt: integer (nullable = true) |-- alm_lvl_higest: integer (nullable = true) |-- vin: string (nullable = true) |-- veh_odo: double (nullable = true) |-- data_batt_subsys_temp_highest_seq: integer (nullable = true) |-- data_batt_sc_volt_lowest_seq: integer (nullable = true) |-- loc_lat84: double (nullable = true) |-- eng_st: integer (nullable = true) |-- veh_soc: integer (nullable = true) |-- data_batt_subsys_volt_highest_seq: integer (nullable = true) |-- eng_spd: integer (nullable = true) |-- alm_common_insulation: integer (nullable = true) |-- alm_common_soc_hop: integer (nullable = true) |-- alm_common_flag: string (nullable = true) |-- alm_common_dcdc_temp: integer (nullable = true) |-- data_batt_temp_probe_lowest_seq: integer (nullable = true) |-- veh_runmode: integer (nullable = true) |-- veh_st: integer (nullable = true) |-- veh_spd: double (nullable = true) |-- data_batt_temp_highest: integer (nullable = true) |-- loc_st: integer (nullable = true) |-- alm_common_sc_low: integer (nullable = true) |-- veh_gear: string (nullable = true) |-- alm_common_dmc_temp: integer (nullable = true) |-- alm_common_sc_high: integer (nullable = true) |-- data_batt_sc_volt_highest_seq: integer (nullable = true) |-- alm_common_hvil_st: integer (nullable = true) |-- dm_data: array (nullable = true) | |-- element: struct (containsNull = true) | | |-- dm_ctl_dc_curr: double (nullable = true) | | |-- dm_ctl_temp: integer (nullable = true) | | |-- dm_ctl_volt: double (nullable = true) | | |-- dm_seq: integer (nullable = true) | | |-- dm_spd: integer (nullable = true) | | |-- dm_st: integer (nullable = true) | | |-- dm_temp: integer (nullable = true) | | |-- dm_torq: integer (nullable = true) 从而构造出我们在 JSON 转换时用到的 schema：

真的无聊， 我在学习 Nim。 Nim 的升级使用 choosenim update stable, choosenim 的升级使用 choosenim update self。

codewars Highest and Lowest In this little assignment you are given a string of space separated numbers, and have to return the highest and lowest number. Example: Kata.HighAndLow("1 2 3 4 5"); // return "5 1" Kata.HighAndLow("1 2 -3 4 5"); // return "5 -3" Kata.HighAndLow("1 9 3 4 -5"); // return "9 -5" Notes: All numbers are valid Int32, no need to validate them. There will always be at least one number in the input string.

roundrobin 在中文里意味着 「设置轮流捡取」。 roundrobin 用于构建 list of lists. say roundrobin <a b c>, <d e f>, <g h i>; # OUTPUT: «((a d g) (b e h) (c f i))␤» say .join(",") for roundrobin([1, 2], [2, 3], [3, 4]); # OUTPUT: «1,2,3␤ # 2,3,4␤» 一旦输入列表中的一个或多个列表中的元素用光了, roundrobin 不会终止, 而是继续直到所有列表的所有元素用光为止。 say roundrobin <a b c>, <d e f m n o p>, <g h i j>; # OUTPUT: «((a d g) (b e h) (c f i) (m j) (n) (o) (p))␤» say .

bettercap 入门 bettercap -autostart net.probe on, set ticker.commands net.show, ticker on bettercap -eval "net.probe on; set ticker.commands net.show; ticker on" help wifi active 显示活动的模块 get wifi.ap.encryption # 获取参数值 get * # 获取所有参数 set wifi.ap.encryption false # 设置参数 help net.recon # 探测模块帮助 wifi.recon on # 开启 wifi 探测 wifi.show # 显示 wifi 列表 wifi.deauth 72:fe:a1:d2:fb:c2 # 攻击一次 全自动 wifi 攻击 help ticker set ticker.commands "wifi.deauth d4:ee:07:67:cc:16" set ticker.period 1 # 设置每秒攻击一次 ticker on ticker off 全自动探测 wifi 热点 set ticker.

Use the exa command line instead of ls to list tree of code: master ~/work/cihon/gac/socket-streaming> exa --tree socket-streaming/src socket-streaming/src └── main ├── resources │ ├── application.conf │ ├── application.prd.conf │ ├── application.test.conf │ ├── log4j.properties │ └── metrics.properties └── scala └── com └── gac └── xs6 └── bigdata ├── BigdataApplication.scala ├── conf │ └── SparkConfiguration.scala ├── core │ ├── Adapter.scala │ └── impl │ ├── AdapterImpl.scala │ └── SubTripImpl.scala ├── model │ ├── SourceData.

编译 SHC 截至目前, Structured Streaming 中的 Sink 还不支持 HBase! Stackoverflow 上有一个解决方案是用第三方的 shc自定义自己的 HBase Sink。有人按照第二个答案说搞不出来, 我也照着做了一遍, 发现是真的! 继续往下翻, 发现 github 上有个人做了一丢丢的修改, 于是我照抄了过来: package org.apache.spark.sql.execution.datasources.hbase import org.apache.spark.sql.{DataFrame, SQLContext, Row} import org.apache.spark.sql.execution.streaming.Sink import org.apache.spark.sql.sources.{DataSourceRegister, StreamSinkProvider} import org.apache.spark.sql.streaming.OutputMode import org.apache.spark.sql.catalyst.CatalystTypeConverters class HBaseSink(options: Map[String, String]) extends Sink with Logging { // String with HBaseTableCatalog.tableCatalog private val hBaseCatalog = options.get("hbasecat").map(_.toString).getOrElse("") override def addBatch(batchId: Long, data: DataFrame): Unit = synchronized { val schema = data.schema val res = data.

谷歌翻译 网易云音乐

While programming in Spark Streaming, I’d like to send some fake data to socket. although the command nc -lk 9999 is useful, but you have to type the data manually! With Raku’s IO::Socket::Async module, I can easily send data to socket continuously. In the server side, I listen a port on 3333, I will print the current time every second, every 5 seconds, skip the report of time: my $vin = 'LSJA0000000000091'; my $last_meter = 0; # 当前里程数 react { whenever IO::Socket::Async.

Use Spark to read and write HBase data 启动 hbase start-hbase.sh 在 HBase 中准备 sample 数据 1、运行 HBase shell hbase shell 2、创建一个含有 Personal 和 Office 列簇的 Contacts 表 create 'Contacts', 'Personal', 'Office' 3、加载一些样例数据 put 'Contacts', '1000', 'Personal:Name', 'John Dole' put 'Contacts', '1000', 'Personal:Phone', '1-425-000-0001' put 'Contacts', '1000', 'Office:Phone', '1-425-000-0002' put 'Contacts', '1000', 'Office:Address', '1111 San Gabriel Dr.' put 'Contacts', '8396', 'Personal:Name', 'Calvin Raji' put 'Contacts', '8396', 'Personal:Phone', '230-555-0191' put 'Contacts', '8396', 'Office:Phone', '230-555-0191' put 'Contacts', '8396', 'Office:Address', '5415 San Gabriel Dr.

Apache HBase Connector 这个 Hbase 连接器可以使用 Spark 读写 Hbase。但是 github 上没有说明怎么把它添加到 pom 文件中。 Google 了一下， 在 Maven 仓库 有这个依赖： <!-- https://mvnrepository.com/artifact/com.hortonworks/shc-core --> <dependency> <groupId>com.hortonworks</groupId> <artifactId>shc-core</artifactId> <version>1.1.1-2.1-s_2.11</version> </dependency> 添加到 pom 文件里面发现还不行， 报红。说明找不到依赖！注意到下面还有行小字： Note: this artifact it located at Hortonworks repository 点击 hortonworks 看下右边的 Indexed Repositories ， 其中每一行对应着仓库配置中的一个 id： Hortonworks， 这个 ID 在下面的仓库配置中就对应 <id>。 这样的 jar 包在标准仓库里面没有，需要你在 pom 文件里面添加上仓库地址： <repository> <id>hortonworks</id> <name>Hortonworks Repository</name> <url>http://repo.hortonworks.com/content/repositories/releases/</url> </repository>

pip install .whl

饭否

百度贴吧

使用Jupyter 打开Spark 的设置方法是： edm shell export PYSPARK_DRIVER_PYTHON=jupyter export PYSPARK_DRIVER_PYTHON_OPTS='notebook' pyspark

根据字节点中的属性值提取父节点 今天使用 Python 的 lxml 模块来提取网页中的内容, 有一个 XPath 的用法不明白, 问题是， table 子节点下有一系列 tr 子节点, 每个 tr 子节点里有 3 列 (td), 如果 td 中 style 属性的值为 color: 0， 那么就不提取它所属的这一 tr 子节点。 查了很久 stackoverflow 才解决, 方法一： from lxml import etree html = ''' <html> <table> <tr id="id_l107" class="new nick nick_cxreg"> <td class="time" ><a href="/raku/2016-09-21>05:07</a></td> <td style="color: #8d8100" class="nick">cxreg</td> <td class="msg">probably useful</a></td> </tr> <tr id="id_l108" class="new nick nick_mask"> <td class="time" ><a href="/raku/2016-09-21>05:09</a></td> <td style="color: #8d8100" class="nick">mask</td> <td class="msg">maybe better directed at <a href="/moarvm/today">#moarvm</a></td> </tr> <tr id="id_l74" class="cont special dark"> <td class="time" ><a href="/raku/2016-09-21>02:37</a></td> <td style="color: 0" class="nick"></td> <td class="msg">mcmillhj joined <a href="/raku/today">#raku</a></td> </tr> <table> </html> ''' selector = etree.

环境搭建

新建工程 scrapy startproject fanfou 该工程的目录结构如下图所示: main.py 文件的内容如下: # -*- coding:utf-8 -*- from scrapy import cmdline cmdline.execute("scrapy crawl fanfou".split()) 这个文件与 scrapy.cfg 存放在同一目录下, 用于在 Pycharm 中调试和运行爬虫时使用, 如果 Pycharm 项目中没有这个文件的话, 会提示找不到模块 fanfou。 在 items.py 中设置要抓取的字段: # -*- coding: utf-8 -*- from scrapy.item import Item, Field class FanfouItem(Item): # define the fields for your item here like: home_url = Field() title = Field() avatar = Field() content = Field() location = Field() Spiders #-*- coding:utf-8 -*- import scrapy from scrapy.

有一个使用场景：将 tar.gz 读成字节数组(byte array) 然后发往 Kafka。我要查看 Kafka 里面的消息, 直接查看的话就是二进制的乱码，首先想到的是把 Kafka 里面保存的字节数组存储到本地，每一行存成一个 .tar.gz 文件： # -*- coding: utf-8 -*- import sys,os from confluent_kafka import Consumer, KafkaError import gzip def run(args): c = Consumer({ "bootstrap.servers" : args[1], # broker 'group.id': args[2], # group id, 每次不一样 'default.topic.config': { 'auto.offset.reset': 'earliest' } }) c.subscribe([ args[3] ]); # topic i=0 while True: msg = c.poll(1.0) if msg is None: continue if msg.error(): if msg.error().code() == KafkaError.

安装 mongodb 到 mongodb 的官网下载后安装。以 Windows 下的使用为例。 在安装路径 C:\Program Files\MongoDB\Server\3.2\bin 目录下, 新建一个名为 data 的文件夹, 同时新建一个 start.bat 批处理文件用于启动 mongodb: @echo off mongod --dbpath ./data 安装 pymongo: pip install pymongo 安装 mongodb 可视化工具 这里测试使用的是 Robomongo 0.9.0-RC10, 体积小, 方便下载。安装完成后打开该软件: 设置好之后保存, 然后点击连接按钮。剩下的就交给代码了。 测试 MongoDB 数据库 新建一个爬虫工程, 在与 scrapy.cfg 同级的目录下新建一个 controlDB.py 文件: import pymongo connecton = pymongo.MongoClient() # 建立连接 test_db = connecton.Daomubiji # 连接上名为 Daomubiji 的数据库 post_info = test_db.test # 获取名为 test 的文档(表) sanshu = {'name': u'盗墓', 'age': 27, 'Skill': 'Python'} mengzhu = {'name': u'张无忌', 'age': 23, 'Skill': '乾坤大挪移', 'girlfriend':'赵敏'} post_info.

setting.py # -*- coding: utf-8 -*- import scrapy_redis BOT_NAME = 'CrawlWithRedis' SPIDER_MODULES = ['CrawlWithRedis.spiders'] NEWSPIDER_MODULE = 'CrawlWithRedis.spiders' ITEM_PIPELINES = {'CrawlWithRedis.pipelines.CrawlWithRedisPipeline':300} # Obey robots.txt rules ROBOTSTXT_OBEY = True # Redis 数据库设置 SCHEDULER = "scrapy_redis.scheduler.Scheduler" SCHEDULER_PERSIST = True SCHEDULER_QUEUE_CLASS = "scrapy_redis.queue.SpiderPriorityQueue" REDIS_URL = None REDIS_HOST = '127.0.0.1' REDIS_PORT = 6379 # MongonDB 设置 MONGODB_HOST = '127.0.0.1' MONGODB_PORT = 27017 MONGODB_DBNAME = 'XiaoYunKeji' MONGODB_DOCNAME = 'daomubiji' items.py # -*- coding: utf-8 -*- # Define here the models for your scraped items # # See documentation in: # http://doc.

Mac 升级到 10.13 之后， sudo -H pip install xxxx 某些模块会出现 Operation not permitted 权限问题: 解决办法是： pip install ipython --user -U 燃鹅儿，安装完后 command not found - WTF！ 反复几次之后，准备放弃之际，又搜索了一下： sudo pip install airflow --upgrade --ignore-installed 竟然解决了！

启动 MongoDB 和 Robomongo 可视化界面。 新建爬虫项目 scrapy startproject mudao 在 scrapy.cfg 文件同级目录中新建一个 main.py, 内容如下: #-*- coding:utf-8 -*- from scrapy import cmdline cmdline.execute("scrapy crawl mudao".split()) settings.py 中增加的额外配置如下: ITEM_PIPELINES = { 'mudao.pipelines.MudaoPipeline': 300, } MONGODB_HOST = '127.0.0.1' MONGODB_PORT = 27017 MONGODB_DBNAME = 'Daomubiji' MONGODB_DOCNAME = 'Book' 这样当 spiders.py 中用到 mongodb 的一些配置时, 导入这个配置文件, 以 seetings['key'] 的形式访问那些配置。 items.py 设置如下: from scrapy import Item, Field class MudaoItem(Item): bookName = Field() bookTitle = Field() chapterNum = Field() chapterName = Field() chapterURL = Field() 总共要提取网页中的 5 个字段, 所以需要 5 个类变量。

Python 3 有一个很好的终端命令录制工具，叫做 asciinema, 使用 pip 安装就好了。它不支持 windows。 pip install asciinema 安装完，开始录制 asciinema rec demo 录制结束后按 Ctrl + D 结束。 查看录制的东西使用 asciinema play demo 分享

我在本系列文章的开头提到，Mojolicious 应用程序不仅仅是 Web 服务器。然后，我展示了如何使用 daemon 守护程序或 prefork 命令启动 Web 服务器。在上一篇文章中，我提到了一个 inflate命令，它可以帮助您将应用程序从 Lite 增加到 Full。 但是，还有其他命令，内置在您的应用程序中，可以帮助您立即提高工作效率！ 命令基础 在开始之前，我想简要讨论与 Mojolicious 发行版捆绑在一起的 mojo 应用程序/脚本。这个命令是一个很小的Mojolicious应用程序（实际上是另一个“hello world”），可以被认为是“空应用程序”。内置命令适用于您的应用程序和此null应用程序，因此请使用更合适的选项。当哪个应用程序运行命令无关紧要时，您可以使用 mojo。 每个命令都带有一行描述和一个（可能是多行）用法语句。要查看可用命令，请运行 mojo help，您将看到所有命令及其说明。你应该看到这样的东西： $ mojo help Usage: APPLICATION COMMAND [OPTIONS] mojo version mojo generate lite_app ./myapp.pl daemon -m production -l http://*:8080 ./myapp.pl get /foo ./myapp.pl routes -v Tip: CGI and PSGI environments can be automatically detected very often and work without commands. Options (for all commands): -h, --help Get more information on a specific command --home <path> Path to home directory of your application, defaults to the value of MOJO_HOME or auto-detection -m, --mode <name> Operating mode for your application, defaults to the value of MOJO_MODE/PLACK_ENV or "development" Commands: cgi Start application with CGI cpanify Upload distribution to CPAN daemon Start application with HTTP and WebSocket server eval Run code against application generate Generate files and directories from templates get Perform HTTP request inflate Inflate embedded files to real files prefork Start application with pre-forking HTTP and WebSocket server psgi Start application with PSGI routes Show available routes test Run tests version Show versions of available modules See 'APPLICATION help COMMAND' for more information on a specific command.

关于Mojolicious最常见的误解之一是，我们目前看到的声明式“Lite”应用程序与大型结构化“完整”应用程序之间存在很大差异。没有东西会离事实很远。 Mojolicious::Lite 是一个非常小的包装器，围绕着所谓的“完整”应用程序架构，赋予它平易近人的关键字语法。 因为文档的简洁单文件示例要好得多，所以 Mojolicious 的大多数文档大多数时候都使用Lite语法。可以理解的是，即使一旦他们的应用程序受益于面向对象的结构，人们也会担心迁移（或者我们称之为“成长”）;毕竟所有的文档似乎都面向Lite应用程序。但是，让这些担忧消失，过渡很容易。一旦你理解了它，文档化的例子就很难翻译了。 此外，Mojolicious 在转换时提供两种形式的帮助。第一个是成长指南，涵盖了这篇文章的所有内容，但是从移植现有应用程序的角度来看（我不会在这里复制）。第二个是 inflate 命令，它甚至可以通过将模板从数据部分移动到自己的文件中来启动您的过程。 也就是说，为了进一步揭开神秘面纱的神秘面纱，我将介绍一些差异，并对Lite语法本身拉开帷幕。 让我说服你 在反复试图说服人们两者之间几乎没有什么区别后，我发现有一种非常好的方式来转变对话。我告诉他们代码。没有真正看一看。在撰写本文时，Mojolicious::Lite 只有37行代码（由David A. Wheeler的 SLOCCount计算的）！ 37 行代码有多大差异？ 好了，现在你相信我，让我们谈谈这些差异。 脚本和类 在 Lite 脚本中，您的应用程序逻辑就位于脚本中。如果是一个完整的应用程序，你的逻辑进入一个单独的类，主要是在 startup 方法中，但删除 app->start 这一行。虽然方法（调用者）的第一个参数通常被称为 $self，你会看到，为了在本系列中保持清晰，我将始终使用 $app。所以我们有： sub startup { my $app = shift; ... # the rest of what was your script } 与此同时，运行的脚本只是启动应用程序的几行。该脚本始终是相同的，与您的应用程序无关，而是要调用的类的名称。我只是使用成长指南末尾的那个。 关键词 现在代码生活在正确的位置，它需要被转换为面向对象。第一步是将逻辑放入一个名为 startup 的方法中，该方法将应用程序对象作为其第一个参数。 实际上有三种类型的关键字，即应用程序对象或应用程序上的方法，路由上的方法和 group。 app 关键字就是之前的调用，因此 app 成为 $app。关键字 helper，hook 和 plugin 只是 app 上的方法，所以 plugin ... 变成 $app->plugin(...) 等。

我们从多年的编程中知道的一件事是，如果你不需要，你就不应该对任何东西进行硬编码。然而，太多的网络应用程序硬编码他们的网址，特别是内部网站。但是，如果你不需要怎么办？ 每个 Mojolicious 路由都有自己的名字，可用于生成网址。如果您没有指定一个，则会生成一个，但您不应该依赖该名称，给它一个有意义且与您的目的相关的名称。在lite应用程序中，名称是任何默认值或回调之后的最后一个参数。 （在完整的应用程序中，它是一个属性，但我们将在另一篇文章中讨论这些内容）。 然后，当您需要一个URL而不是硬编码时，使用 url_for 或相关功能按名称生成 URL，您甚至可以根据需要传递占位符值。让我们看看它是如何工作的！ 北极主页 圣诞老人从一个简单的 webapp 开始。它只有一个主页和一个页面供他和他的员工使用。由于它如此简单，他没有任何更深的路径，只有 / 用于主页和 /:name 用于员工页面。当然，我要告诉你的是一个简化，我无法向你展示圣诞老人的完整网站，原因很明显。 use Mojolicious::Lite; get '/:name' => {template => 'staff'} => 'staff'; get '/' => {template => 'home'} => 'home'; app->start; __DATA__ @@ staff.html.ep <p>This is <%= ucfirst $name %>.</p> @@ home.html.ep <p>Welcome to The North Pole!</p> <p> Say hi to <%= link_to 'Santa' => staff => {name => 'santa'} %> and <%= link_to 'Rudolph' => staff => {name => 'rudolph'} %>.

在 Mojolicious 中，当处理请求并准备响应时，最重要的概念之一是“藏匿”。由于它是一个非阻塞框架，因此您的代码在处理期间不能使用全局变量来存储任何状态。如果您运行了一些其他代码，那么很容易在请求之间进行串扰。 存储是您在处理信息时可以存储信息的地方。它只是一个简单的哈希引用，它附加到处理请求的控制器对象上。它与这一笔交易一起生存和死亡。 虽然你可以而且应该将它用作暂存器，但实际上还是要多得多。存储控制几乎控制了您生成的响应的每个方面。让我们仔细看看它是如何工作的 使用 Stash 渲染文本 在上一篇文章中，我们讨论了最简单的“Hello world”应用程序。 use Mojolicious::Lite; get '/' => {text => 'Hello 🌍 World!'}; app->start; 虽然这是一个非常简单的工作案例，但更常见的例子就是这样 use Mojolicious::Lite; get '/' => sub { my $c = shift; $c->render(text => 'Hello 🌍 World!'); }; app->start; 在此示例中，GET /请求由“动作回调”处理。回调是函数引用，打算将来调用;在这种情况下，当客户端请求进入与该类型的请求匹配时，将调用回调。 使用一个参数调用操作，称为控制器。控制器是一个对象，表示我们的应用程序与当前事务的交互。它包含一个表示事务的对象，该对象又保存请求和响应的对象。它有一些方法可用于生成响应，其中一个是渲染，您可以在上面看到。在这里，您会看到我们将呈现一些文本。 事实上，渲染的大多数参数实际上只是合并到了藏匿处。实际上，上面的例子是相同的 use Mojolicious::Lite; get '/' => sub { my $c = shift; $c->stash(text => 'Hello 🌍 World!'); $c->render; }; app->start; 你现在看到的是，Mojolicious 看着藏匿处看看如何呈现回应。事实上，如果你不调用 render，但它有足够的信息来在 stash 中呈现响应，它就会这样做。 use Mojolicious::Lite; get '/' => sub { my $c = shift; $c->stash(text => 'Hello 🌍 World!

从头开始 在这个出现日历系列中，一些帖子将是介绍性的，一些将是高级的，一些将是新功能。谁知道接下来会发生什么？但是现在让我们通过研究如何开始来确保公平竞争。 什么是 Mojolicious？ 嗯，Mojolicious 真的是两件事。首先，它是一个强大的以 Web 为重点的工具包，名为 Mojo。其次，它是一个名为 Mojolicious 的强大的 Web 框架。 Mojolicious 框架是使用 Mojo 工具包构建的。 这并不意味着你不能在其他地方使用 Mojo 工具。如果您看到一些您喜欢的工具但想要与其他框架一起使用，请继续，我不会告诉您！在任何你想要的 Perl 代码中使用它！ 拍我不会提到 Perl！是的，Mojolicious 是用 Perl 编写的。不要让那吓到你。它有很多方法可以保证您和您的代码安全！从内置对象系统到具有可链接方法的一致 API，Mojolicious 旨在使您的代码保持清洁和可读。希望你甚至可以使用它玩得开心！ 安装 安装简单快捷。实际上，如果您将测试工具设置为并行运行，则应在几秒钟内完成安装！您可以通过在您的环境中设置 HARNESS_OPTIONS = j9（其中 9 比您的核心数多一个）来实现。 最简单的安装方法是运行 curl -L https://cpanmin.us | perl - -M https://cpan.metacpan.org -n Mojolicious 或者，您可以使用任何cpan客户端（我们喜欢 cpanm）或使用系统的包管理器进行安装。 你的第一个应用 在编程的传统中，我们需要做的第一件事就是运行一个hello world应用程序。 将以下内容保存为 hello.pl use Mojolicious::Lite; get '/' => {text => 'Hello 🌍 World!'}; app->start; 这个脚本很简单 导入 Mojolicious（精简版） 定义一个 GET 处理程序来响应具有 unicode 版本的 hello world 的请求 启动应用程序 但在此之前，我们必须启动一个 Web 服务器。

替换文件名 使用 replaceRE title: "{{ .TranslationBaseName | replaceRE "^[0-9]{4}-[0-9]{2}-[0-9]{2}-?(.*)" "$1" | title }}" 设置分页 Hugo 不能禁用分页，所以设置一个很大的数（达到 9999 篇文章后分页）： vi config.toml： paginate = 9999

package allinone import org.apache.spark.streaming._ import org.apache.log4j.{Level, Logger} import org.apache.spark.sql.SparkSession /** * Counts words cumulatively in UTF8 encoded, '\n' delimited text received from the network every * second starting with initial value of word count. * Usage: StatefulNetworkWordCount <hostname> <port> * <hostname> and <port> describe the TCP server that Spark Streaming would connect to receive * data. * * To run this on your local machine, you need to first run a Netcat server * `$ nc -lk 9999` * and then run the example * `spark-submit.

测试 Kafka 连通性 往 Kafka 里面写数据时, 默认在不存在那个topic 时，会自动创建 测试能否连上某台机器的 kafka yum install telnet yum install telnet-server telnet 10.10.10.35 9092 如果 Trying 10.10.10.35... Connected to 10.10.10.35. Escape character is '^]'. 说明可以连通。 同样适用于 zoomkeeper: telnet 10.0.201.83 2181 Trying 10.0.201.83... Connected to 10.0.201.83. Escape character is '^]'. 从指定位置消费 Kafka pip install kafka-python import gzip from kafka import KafkaConsumer from kafka import TopicPartition consumer = KafkaConsumer(bootstrap_servers='10.30.10.15:9092') partition = TopicPartition('dc-diagnostic-report', 0) start = 8833 end = 8835 consumer.

你不知道的 JavaScript

函数JavaScript 的函数是一种对象。 对象 Javascript 中没有类这样的语言结构, Javascript 中的对象是一个 名称与值配对的集合。这样一对儿名称和值的配对被称为属性。例如一个人的属性有: 身高: 178cm 体重 65kg 年龄 28 所以, Javascript 对象可以定义为属性 的集合。Javascript 的对象字面量： // 对象字面量表达式的语法 { 属性名 : 属性值, 属性名 : 属性值, ...... } 属性名可以是标识符。字符串和数值： // 对象字面量表达式的例子 { x: 2, y:1 } // 属性名是标识符 { "x":2, "y":1 } // 属性名是字符串值 { 'x':2, 'y':1 } // 属性名是字符串值 { 1:2, 2:1 } // 属性名是数值 { x:2, y:1, enable:true, color:{ r:255, g:255, b:255 } } // 各种类型的属性值 对对象字面量表达式求值所得到的结果，是所生成对象的一个引用。

100-days-of-code-log 100-days-of-code-log-r2

用 parquet 数据模拟实时数据流 import ohmysummer.conf.{KafkaConfiguration, UriConfiguration} import ohmysummer.pipeline.schema.{EnterpriseSchema, NationSchema} import org.apache.spark.sql._ import org.apache.spark.sql.streaming.Trigger import org.apache.spark.sql.functions.{col, struct, to_json} /** * 读取 parquet 文件转为 JSON 后写到 HDFS, 在用命令行将 JSON 数据逐行发到 Kakfa 模拟实时流 */ object WriteEnterprise2Kafka { def main(args: Array[String]): Unit = { val spark: SparkSession = SparkSession .builder .master("local[2]") .appName("Write Enterprise Parquet to Kafka") .getOrCreate() val parquetSchema = (new EnterpriseSchema).schema val parqurtUri = (new UriConfiguration).xs6enterprise val topics = (new KafkaConfiguration).topics val bootstrap_servers = (new KafkaConfiguration).

环境变量

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>mileageanxiety</groupId> <artifactId>wmanxiety</artifactId> <version>1.0-SNAPSHOT</version> <packaging>jar</packaging> <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <!-- Languages --> <java.version>1.8</java.version> <scala.version>2.11.8</scala.version> <scala.binary.version>2.11</scala.binary.version> <!-- Apache Spark --> <spark.version>2.1.0</spark.version> <hadoop.version>2.6.0-cdh5.13.2</hadoop.version> <!-- Thirty Party --> <scopt.version>3.3.0</scopt.version> <typesafe-config.version>1.3.0</typesafe-config.version> <hbase.version>1.1.2</hbase.version> </properties> <dependencies> <!-- https://mvnrepository.com/artifact/com.hortonworks/shc-core --> <dependency> <groupId>com.hortonworks</groupId> <artifactId>shc-core</artifactId> <version>1.1.1-2.1-s_2.11</version> </dependency> <!--<dependency>--> <!--<groupId>org.apache.spark</groupId>--> <!--<artifactId>spark-core_2.11</artifactId>--> <!--<version>2.1.0</version>--> <!--</dependency>--> <!--<dependency>--> <!--<groupId>org.apache.spark</groupId>--> <!--<artifactId>spark-sql_2.10</artifactId>--> <!--<version>2.1.0</version>--> <!--</dependency>--> <!--<dependency>--> <!--<groupId>org.scala-lang</groupId>--> <!--<artifactId>scala-library</artifactId>--> <!--<version>2.11.8</version>--> <!--</dependency>--> <!-- https://mvnrepository.com/artifact/com.typesafe/config --> <dependency> <groupId>com.typesafe</groupId> <artifactId>config</artifactId> <version>1.3.0</version> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.phoenix/phoenix-spark --> <!

Kafka Test Environment

PySpark

在 IDEA 中配置 Pyspark 参考 https://had00ping.wordpress.com/2017/11/14/setting-intellij-for-pyspark/ 最后添加根目录： 结果如下： 在 IDEA 中添加 java 目录 删除 pom 文件中的如下配置: <sourceDirectory>src/main/scala</sourceDirectory> <sourceDirectory>src/main/java</sourceDirectory>

mill 是一个新的 build 工具, 可以在命令行中导入 jar 包: import $ivy.`com.alibaba:fastjson:1.2.12` import com.alibaba.fastjson.{JSON, JSONObject} val j = JSON.parseObject("{'a':123}") println(j.get("a")) // 123

例如，某个依赖在 pom 文件中是这样的： <dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>2.6.2</version> </dependency> 那么使用 mvn 单独下载这个依赖： mvn org.apache.maven.plugins:maven-dependency-plugin:2.1:get \ -DrepoUrl=https://mvnrepository.com/artifact/ -Dartifact=redis.clients:jedis:2.6.2

https://github.com/scopt/scopt 简单的 scala 命令行选项解析 scopt 是一个小小的命令行选项解析库。 Sonatype libraryDependencies += "com.github.scopt" %% "scopt" % "X.Y.Z" 查看上面的 Maven Central badge 使用方法 scopt 提供了两种解析方式：immutable 和 mutable。无论哪种情况，首先您需要一个表示配置的 case class： import java.io.File case class Config(foo: Int = -1, out: File = new File("."), xyz: Boolean = false, libName: String = "", maxCount: Int = -1, verbose: Boolean = false, debug: Boolean = false, mode: String = "", files: Seq[File] = Seq(), keepalive: Boolean = false, jars: Seq[File] = Seq(), kwargs: Map[String,String] = Map()) 在不可变的解析样式中，config 配置对象作为参数传递给 action 回调。另一方面，在可变解析样式中，你需要修改配置对象。

大多数情况下， Scala 中的 trait 相当于 Java 中的借口， 或者 Raku 中的 Role。Scala 可以继承多个 trait。 trait 作为接口 trait BaseSoundPlayer { def play def close def pause def stop def resume } 和 OC 中的接口类似， 如果方法带有参数，则声明时加上参数即可： trait Dog { def speak(whatToSay: String) def wagTail(enabled: Boolean) } 类继承 trait 时需要使用 extends 和 with 关键字, 如果类只继承一个 trait, 则只使用 extends 就够了： class Mp3SoundPlayer extends BaseSoundPlayer { ... 继承一个类和一个或多个 trait 时，对类使用 extends, 对 trait 使用 with： class Foo extends BaseClass with Trait1 with Traits { .

package ohmysummer import java.sql.Timestamp import org.apache.spark.sql.SparkSession import org.apache.spark.sql.streaming._ /** * Counts words in UTF8 encoded, '\n' delimited text received from the network. * * Usage: MapGroupsWithState <hostname> <port> * <hostname> and <port> describe the TCP server that Structured Streaming * would connect to receive data. * * To run this on your local machine, you need to first run a Netcat server * `$ nc -lk 9999` * and then run the example * `$ bin/run-example sql.

package ohmysummer import ohmysummer.model.SourceCan import ohmysummer.pipeline.kafka.WmKafkaDeserializer import ohmysummer.pipeline.schema.{CanSignalSchema, DcSaleData} import org.apache.spark.sql.{DataFrame, Dataset, SparkSession} import org.apache.spark.sql.streaming.OutputMode import org.apache.spark.sql.functions._ /** * 从 Kafka 读取 JSON 数据 * https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html * https://stackoverflow.com/questions/43297973/how-to-read-records-in-json-format-from-kafka-using-structured-streaming * https://stackoverflow.com/questions/48361177/spark-structured-streaming-kafka-convert-json-without-schema-infer-schema * https://databricks.com/blog/2017/04/26/processing-data-in-apache-kafka-with-structured-streaming-in-apache-spark-2-2.html * https://databricks.com/blog/2017/02/23/working-complex-data-formats-structured-streaming-apache-spark-2-1.html */ object WmDcSaleApplication { def main(args: Array[String]) { val spark = SparkSession .builder .appName("ReadFromKafka") .master("local[2]") .getOrCreate() object KafkaDeserializerWrapper { val deser = new WmKafkaDeserializer } spark.udf.register("deserialize", (topic: String, bytes: Array[Byte]) => KafkaDeserializerWrapper.deser.deserialize(topic, bytes) ) val df = spark.

重启 hdfs 集群后，出现的standy 错误的原因: 是因为没有启动zookeeper, zookeeper不会自动重启, zook的启动命令是zkServer.sh. 以后启动集群时，先启动那slave上安装了 zookeeper 的 zookeeper , 然后再启动 hdfs。 Master 上 的 zoomkeeper, vim /opt/zookeeper-3.4.9/conf/zoom.cfg: # The number of milliseconds of each tick tickTime=2000 # The number of ticks that the initial # synchronization phase can take initLimit=10 # The number of ticks that can pass between # sending a request and getting an acknowledgement syncLimit=5 # the directory where the snapshot is stored. # do not use /tmp for storage, /tmp here is just # example sakes.

以下所有操作都是在 Master 上。 停掉所有的 spark crontab 任务 stop crontab sudo rm /etc/cron.d/crontab_ald_spark 杀掉所有的正在运行的 spark 进程 kill spark progress ps aux | grep 'spark' | awk '{print $2}' | xargs kill -9 杀掉 Master(10.0.0.247) stop-master.sh spark/sbin/stop-master.sh -h 10.0.0.247 -p 7077 停掉所有的 slaves stop-slaves.sh spark/sbin/stop-slaves.sh -h 10.0.0.247 -p 7077 停掉 slaves, 注意是 s, 需要先做 Master 到各 slave 的免密码登陆, 即把 Master 下的 ~/.ssh/id_rsa.pub 中的公钥复制到各 slave 下(用sudo su切换到root)的 ~/.ssh/authorized_keys 中。 在 Master 下的 /etc/hosts 文件中, 加上各 slave 的主机名:

salve 上, root@slave2:/opt/spark-2.1.0-bin-hadoop2.6/conf# vim spark-env.sh: export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-amd64 export SCALA_HOME=/opt/scala-2.11.8 export SPARK_MASTER_IP=namenode export HADOOP_HOME=/opt/hadoop-2.6.5 export SPARK_WORKER_CORES=3 export SPARK_WORKER_MEMORY=12g export HADOOP_CONF_DIR=/opt/hadoop-2.6.5/etc/hadoop export HADOOP_CONF_LIB_NATIVE_DIR=/opt/hadoop-2.6.5/lib/native export HADOOP_MAPRED_HOME=/opt/hadoop-2.6.5 export HADOOP_COMMON_HOME=/opt/hadoop-2.6.5 export HADOOP_HDFS_HOME=/opt/hadoop-2.6.5 export YARN_HOME=/opt/hadoop-2.6.5 export HADOOP_INSTALL=/opt/hadoop-2.6.5 export YARN_CONF_DIR=/opt/hadoop-2.6.5/etc/hadoop export SPARK_HOME=/opt/spark-2.1.0-bin-hadoop2.6 export SPARK_CLASSPATH=/opt/hadoop-2.6.5/etc/hadoop:/opt/hadoop-2.6.5/share/hadoop/common/lib/*:/opt/hadoop-2.6.5/share/hadoop/common/*:/opt/hadoop-2.6.5/share/hadoop/hdfs:/opt/hadoop-2.6.5/share/hadoop/hdfs/lib/*:/opt/hadoop-2.6.5/share/hadoop/hdfs/*:/opt/hadoop-2.6.5/share/hadoop/yarn/lib/*:/opt/hadoop-2.6.5/share/hadoop/yarn/*:/opt/hadoop-2.6.5/share/hadoop/mapreduce/lib/*:/opt/hadoop-2.6.5/share/hadoop/mapreduce/*:/opt/hadoop-2.6.5/contrib/capacity-scheduler/*.jar slave2上 yarn-site.xml <?xml version="1.0"?> <!-- Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.

Spark 中的 –files 参数与 ConfigFactory 工厂方法 scala 对象 以前有个大数据项目做小程序统计，读取 HDFS 上的 Parquet 文件，统计完毕后，将结果写入到 MySQL 数据库。首先想到的是将 MySQL 的配置写在代码里面: val jdbcUrl = "jdbc:mysql://127.0.0.1:6606/test?useUnicode=true&characterEncoding=utf-8&autoReconnect=true&failOverReadOnly=false&useSSL=false" val user = "root" val password = "averyloooooongword" val driver = "com.mysql.jdbc.Driver" properties 文件 如果是测试，生产环境各有一套，那上面的代码就要分别复制俩份，不便于维护！后来知道了可以把配置放在 resources 目录下, 针对本地，测试和生产环境，分别创建不同的 properties 文件： conf.properties conf_product.properties env.properties local.properties 例如其中的 conf.properties 内容如下： # 测试环境配置 ## 数据库配置 jdbc.url=jdbc:mysql://10.0.0.11:3306/ald_xinen_test?useUnicode=true&characterEncoding=utf-8&autoReconnect=true&failOverReadOnly=false jdbc.user=aldwx jdbc.pwd=123456 jdbc.driver=com.mysql.jdbc.Driver # parquet 文件目录 tongji.parquet=hdfs://10.0.0.212:9000/ald_log_parquet 然后在代码里面读取 resource 文件中的配置： /** * 根据 key 获取 properties 文件中的 value * @param key properties 文件中等号左边的键 * @return 返回 properties 文件中等号右边的值 */ public static String getProperty(String key) { Properties properties = new Properties(); InputStream in = ConfigurationUtil.

看一下提交命令 offline.sh 中的一个有趣的配置： spark2-submit \ --class $1 \ --master yarn \ --deploy-mode cluster \ --driver-memory 4g \ --driver-cores 2 \ --executor-memory 6g \ --executor-cores 3 \ --num-executors 12 \ --conf spark.yarn.submit.waitAppCompletion=false \ --files /etc/hbase/conf/hbase-site.xml \ /tmp/xxxx.jar spark.yarn.submit.waitAppCompletion 这个配置，看字面意思，提交任务，直到程序结束运行的意思。如果设置为 false， 那么提交完就可以去干别的事情了，不用一直等着看结果；如果设置为 true（默认的）， 顾名思义，提交完程序后会一直在终端中打印信息，直到程序运行结束。所以为了方便，还是设置成 false 比较合适。因为等一会你可以通过 yarn logs applicationId application_1534402443030_0345 查看运行的日志/运行结果。 ./offline.sh wmstat.Completion 12 7 它提交完就完事了，任务还在后台运行，所以打印出来的日志很短： SLF4J: Class path contains multiple SLF4J bindings. SLF4J: Found binding in [jar:file:/opt/cloudera/parcels/SPARK2-2.3.0.cloudera2-1.cdh5.13.3.p0.316101/lib/spark2/jars/phoenix-4.14.0-cdh5.13.2-client.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J: Found binding in [jar:file:/opt/cloudera/parcels/CDH-5.

数据样例 有两个文件，一个是 json： a.json { "createTime": 1532598069, "event": { "info": { "AAA": "one", "BBB": "two", "DDD": "opps" } } } 另一个也是 json： b.json { "createTime": "1532598069", "event": { "info": { "AAA": "three", "BBB": "four", "CCC": "haha" } } } Kafka Producer info 里面的字段个数是不固定的。用下面的代码先将 a.json 发送到 Kafka： from confluent_kafka import Producer p = Producer({'bootstrap.servers': 'localhost:9092'}) def delivery_report(err, msg): """ Called once for each message produced to indicate delivery result. Triggered by poll() or flush().

Producer 发送 JSON 数据到 Kafka： from confluent_kafka import Producer p = Producer({'bootstrap.servers': 'localhost:9092'}) def delivery_report(err, msg): """ Called once for each message produced to indicate delivery result. Triggered by poll() or flush(). """ if err is not None: print('Message delivery failed: {}'.format(err)) else: print('Message delivered to {} [{}]'.format(msg.topic(), msg.partition())) with open("/Users/ohmycloud/bigdata/Spark/camelia.json") as f: data = f.read() p.poll(0) p.produce('net-logs', data.encode('utf-8'), callback=delivery_report) p.flush() JSON 数据内容如下： { "metadata":{ "access_token":"c.FmDPkzyzaQe...", "client_version":1 }, "devices":{ "thermostats":{ "peyiJNo0IldT2YlIVtYaGQ":{ "device_id":"peyiJNo0IldT2YlIVtYaGQ", "locale":"en-US", "software_version":"4.

在写一个大数据项目的时候，用的 json schema 嵌套了很多层，使用了很多字段。编译的时候就爆出了这个堆栈溢出： scalac: Error: org.jetbrains.jps.incremental.scala.remote.ServerException java.lang.StackOverflowError 解决方法一： 在 pom 文件里面添加 jvmArg， 将 -Xss 设置为 4096K： <configuration> <args> <!--<arg>-make:transitive</arg>--> <arg>-dependencyfile</arg> <arg>${project.build.directory}/.scala_dependencies</arg> </args> <jvmArgs> <jvmArg>-Xss4096K</jvmArg> </jvmArgs> </configuration> 这样 submit 的时候就可以跑了： spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0 --class ohmysummer.SparkStructuredStreaming --master local[2] --driver-memory 2g --driver-cores 2 --executor-memory 2g --executor-cores 2 --num-executors 2 target/wordcount-1.0-SNAPSHOT.jar 解决方法二： 点击 run 按钮的时候仍旧报堆栈溢出， 在 IDEA 中设置：

在浏览器中运行 Scala scalafiddle.io 表达式 表达式是可计算的语句。 你可以使用 println 输出表达式的值。 println(1) // 1 println(1+1) // 2 println("Hello!") // Hello! println("Hello," + " world!") // Hello, world! 值 使用 val 关键字来命名表达式的值。 val x = 1 + 1 println(x) // 2 这里的命名结果, 例如 x, 叫做值。引用一个值不会重新计算它。 值一旦确定就不能被重新赋值。 val x = 1+1 x = 3 // This does not compile. 值的类型可以被推断出来, 但是你也可以显式的声明它的类型, 像这样: val x: Int = 1 + 1 变量 变量像值一样, 但是你可以给它重新赋值。用 var 关键字来定义一个变量。 var x = 1 + 1 x = 3 // This compiles because "x" is declared with the "var" keyword.

升级了一下 IDEA， 运行 spark 程序的时候给我报错了， NoClassDefFound : Scala/xml/metadata 解决方法， 在pom 文件里面添加如下依赖： <dependency> <groupId>org.scala-lang</groupId> <artifactId>scala-library</artifactId> <version>2.11.8</version> </dependency> <!-- https://mvnrepository.com/artifact/org.scala-lang.modules/scala-xml --> <dependency> <groupId>org.scala-lang.modules</groupId> <artifactId>scala-xml_2.11</artifactId> <version>1.1.0</version> </dependency> CDH 集群，Spark 升级到 2.3 后，运行任务报错： Error: A JNI error has occurred, please check your installation and try again Exception in thread "main" java.lang.NoClassDefFoundError: org/slf4j/Logger at java.lang.Class.getDeclaredMethods0(Native Method) at java.lang.Class.privateGetDeclaredMethods(Class.java:2701) at java.lang.Class.privateGetMethodRecursive(Class.java:3048) at java.lang.Class.getMethod0(Class.java:3018) at java.lang.Class.getMethod(Class.java:1784) at sun.launcher.LauncherHelper.validateMainClass(LauncherHelper.java:544) at sun.launcher.LauncherHelper.checkAndLoadMain(LauncherHelper.java:526) Caused by: java.lang.ClassNotFoundException: org.slf4j.Logger at java.net.URLClassLoader.findClass(URLClassLoader.java:381) at java.