- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
尝试在 sparklyr 中拆分一个字符串,然后将其用于连接/过滤
我尝试了将字符串标记化然后将其分离到新列的建议方法。这是一个可重现的示例(请注意,我必须将在 copy_to 之后变成字符串“NA”的 NA 翻译成实际的 NA,有没有办法不必这样做)
x <- data.frame(Id=c(1,2,3,4),A=c('A-B','A-C','A-D',NA))
df <- copy_to(sc,x,'df')
df %>% mutate(A = ifelse(A=='NA',NA,A)) %>% ft_regex_tokenizer(input.col="A", output.col="B", pattern="-",to_lower_case=F) %>%
sdf_separate_column("B", into=c("C", "D")) %>% filter(C=='A')
问题是,如果我尝试过滤新创建的列(例如 %>% filter(C=='A')
或加入它们,我会收到错误,见下文
Error : org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 367.0 failed 4 times, most recent failure: Lost task 0.3 in stage 367.0 (TID 5062, 10.139.64.4, executor 0): org.apache.spark.SparkException: Failed to execute user defined function($anonfun$createTransformFunc$2: (string) => array<string>)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$11$$anon$1.hasNext(WholeStageCodegenExec.scala:622)
at org.apache.spark.sql.execution.collect.UnsafeRowBatchUtils$.encodeUnsafeRows(UnsafeRowBatchUtils.scala:51)
at org.apache.spark.sql.execution.collect.Collector$$anonfun$2.apply(Collector.scala:148)
at org.apache.spark.sql.execution.collect.Collector$$anonfun$2.apply(Collector.scala:147)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.doRunTask(Task.scala:139)
at org.apache.spark.scheduler.Task.run(Task.scala:112)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$13.apply(Executor.scala:497)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1432)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:503)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.NullPointerException
at java.util.regex.Matcher.getTextLength(Matcher.java:1283)
at java.util.regex.Matcher.reset(Matcher.java:309)
at java.util.regex.Matcher.<init>(Matcher.java:229)
at java.util.regex.Pattern.matcher(Pattern.java:1093)
at java.util.regex.Pattern.split(Pattern.java:1206)
at java.util.regex.Pattern.split(Pattern.java:1273)
at scala.util.matching.Regex.split(Regex.scala:526)
at org.apache.spark.ml.feature.RegexTokenizer$$anonfun$createTransformFunc$2.apply(Tokenizer.scala:144)
at org.apache.spark.ml.feature.RegexTokenizer$$anonfun$createTransformFunc$2.apply(Tokenizer.scala:141)
... 15 more
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:2100)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2088)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2087)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2087)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:1076)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:1076)
at scala.Option.foreach(Option.scala:257)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1076)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2319)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2267)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2255)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:873)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2252)
at org.apache.spark.sql.execution.collect.Collector.runSparkJobs(Collector.scala:259)
at org.apache.spark.sql.execution.collect.Collector.collect(Collector.scala:269)
at org.apache.spark.sql.execution.collect.Collector$.collect(Collector.scala:69)
at org.apache.spark.sql.execution.collect.Collector$.collect(Collector.scala:75)
at org.apache.spark.sql.execution.ResultCacheManager.getOrComputeResult(ResultCacheManager.scala:497)
at org.apache.spark.sql.execution.CollectLimitExec.executeCollectResult(limit.scala:48)
at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$collectResult(Dataset.scala:2827)
at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$collectFromPlan(Dataset.scala:3439)
at org.apache.spark.sql.Dataset$$anonfun$collect$1.apply(Dataset.scala:2794)
at org.apache.spark.sql.Dataset$$anonfun$collect$1.apply(Dataset.scala:2794)
at org.apache.spark.sql.Dataset$$anonfun$54.apply(Dataset.scala:3423)
at org.apache.spark.sql.execution.SQLExecution$$anonfun$withCustomExecutionEnv$1.apply(SQLExecution.scala:99)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:228)
at org.apache.spark.sql.execution.SQLExecution$.withCustomExecutionEnv(SQLExecution.scala:85)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:158)
at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$withAction(Dataset.scala:3422)
at org.apache.spark.sql.Dataset.collect(Dataset.scala:2794)
at sparklyr.Utils$.collect(utils.scala:200)
at sparklyr.Utils.collect(utils.scala)
at sun.reflect.GeneratedMethodAccessor577.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at sparklyr.Invoke.invoke(invoke.scala:139)
at sparklyr.StreamHandler.handleMethodCall(stream.scala:123)
at sparklyr.StreamHandler.read(stream.scala:66)
at sparklyr.BackendHandler.channelRead0(handler.scala:51)
at sparklyr.BackendHandler.channelRead0(handler.scala:4)
at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:102)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:284)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1359)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:935)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:138)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:138)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.spark.SparkExcepti
In addition: Warning messages:
1: The parameter `input.col` is deprecated and will be removed in a future release. Please use `input_col` instead.
2: The parameter `output.col` is deprecated and will be removed in a future release. Please use `output_col` instead
不确定为什么创建的列的类型根据 sdf_schema 是“StringType”。
是否有使用 sparklyr 实际分离列的解决方案,我以后可以将其用作字符串,而不必将框架写到文件中,或者不必收集到驱动程序节点?
最佳答案
在这里使用 Spark ML 转换器不是一个好的选择。相反,你应该 split
函数:
df %>%
mutate(B = split(A, "-")) %>%
sdf_separate_column("B", into = c("C", "D")) %>%
filter(C %IS NOT DISTINCT FROM% "A")
# Source: spark<?> [?? x 5]
Id A B C D
<dbl> <chr> <list> <chr> <chr>
1 1 A-B <list [2]> A B
2 2 A-C <list [2]> A C
3 3 A-D <list [2]> A D
或regexp_extract
pattern <- "^(.*)-(.*)$"
df %>%
mutate(
C = regexp_extract(A, pattern, 1),
D = regexp_extract(A, pattern, 2)
) %>%
filter(C %IS NOT DISTINCT FROM% "A")
# Source: spark<?> [?? x 4]
Id A C D
<dbl> <chr> <chr> <chr>
1 1 A-B A B
2 2 A-C A C
3 3 A-D A D
尽管如此,如果你想让 RegexpTokenzier
工作,你必须先处理 NULL
(外部 R 类型中的 NA
)。例如可以使用 coalesce
tokenizer <- ft_regex_tokenizer(
sc, input_col = "A", output_col = "B",
pattern = "-", to_lower_case = F
)
df %>%
mutate(A = coalesce(A, "")) %>%
ml_transform(tokenizer, .) %>%
sdf_separate_column("B", into=c("C", "D")) %>%
filter(C %IS NOT DISTINCT FROM% "A")
# Source: spark<?> [?? x 5]
Id A B C D
<dbl> <chr> <list> <chr> <chr>
1 1 A-B <list [2]> A B
2 2 A-C <list [2]> A C
3 3 A-D <list [2]> A D
或者先删除丢失的数据:
df %>%
# or filter(!is.na(A))
na.omit(columns=c("A")) %>%
ml_transform(tokenizer, .) %>%
sdf_separate_column("B", into=c("C", "D")) %>%
filter(C %IS NOT DISTINCT FROM% "A")
* Dropped 1 rows with 'na.omit' (4 => 3)
# Source: spark<?> [?? x 5]
Id A B C D
<dbl> <chr> <list> <chr> <chr>
1 1 A-B <list [2]> A B
2 2 A-C <list [2]> A C
3 3 A-D <list [2]> A D
关于r - Sparklyr 拆分字符串(到字符串),我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/54522707/
如何使用 SPListCollection.Add(String, String, String, String, Int32, String, SPListTemplate.QuickLaunchO
我刚刚开始使用 C++ 并且对 C# 有一些经验,所以我有一些一般的编程经验。然而,似乎我马上就被击落了。我试过在谷歌上寻找,以免浪费任何人的时间,但没有结果。 int main(int argc,
这个问题已经有答案了: In Java 8 how do I transform a Map to another Map using a lambda? (8 个回答) Convert a Map>
我正在使用 node + typescript 和集成的 swagger 进行 API 调用。我 Swagger 提出以下要求 http://localhost:3033/employees/sear
我是 C++ 容器模板的新手。我收集了一些记录。每条记录都有一个唯一的名称,以及一个字段/值对列表。将按名称访问记录。字段/值对的顺序很重要。因此我设计如下: typedef string
我需要这两种方法,但j2me没有,我找到了一个replaceall();但这是 replaceall(string,string,string); 第二个方法是SringBuffer但在j2me中它没
If string is an alias of String in the .net framework为什么会发生这种情况,我应该如何解释它: type JustAString = string
我有两个列表(或字符串):一个大,另一个小。 我想检查较大的(A)是否包含小的(B)。 我的期望如下: 案例 1. B 是 A 的子集 A = [1,2,3] B = [1,2] contains(A
我有一个似乎无法解决的小问题。 这里...我有一个像这样创建的输入... var input = $(''); 如果我这样做......一切都很好 $(this).append(input); 如果我
我有以下代码片段 string[] lines = objects.Split(new string[] { "\r\n", "\n" }, StringSplitOptions.No
这可能真的很简单,但我已经坚持了一段时间了。 我正在尝试输出一个字符串,然后输出一个带有两位小数的 double ,后跟另一个字符串,这是我的代码。 System.out.printf("成本:%.2
以下是 Cloud Firestore 列表查询中的示例之一 citiesRef.where("state", ">=", "CA").where("state", "= 字符串,我们在Stack O
我正在尝试检查一个字符串是否包含在另一个字符串中。后面的代码非常简单。我怎样才能在 jquery 中做到这一点? function deleteRow(locName, locID) { if
这个问题在这里已经有了答案: How to implement big int in C++ (14 个答案) 关闭 9 年前。 我有 2 个字符串,都只包含数字。这些数字大于 uint64_t 的
我有一个带有自定义转换器的 Dozer 映射: com.xyz.Customer com.xyz.CustomerDAO customerName
这个问题在这里已经有了答案: How do I compare strings in Java? (23 个回答) 关闭 6 年前。 我想了解字符串池的工作原理以及一个字符串等于另一个字符串的规则是
我已阅读 this问题和其他一些问题。但它们与我的问题有些无关 对于 UILabel 如果你不指定 ? 或 ! 你会得到这样的错误: @IBOutlet property has non-option
这两种方法中哪一种在理论上更快,为什么? (指向字符串的指针必须是常量。) destination[count] 和 *destination++ 之间的确切区别是什么? destination[co
This question already has answers here: Closed 11 years ago. Possible Duplicates: Is String.Format a
我有一个Stream一个文件的,现在我想将相同的单词组合成 Map这很重要,这个词在 Stream 中出现的频率. 我知道我必须使用 collect(Collectors.groupingBy(..)
我是一名优秀的程序员,十分优秀!