这个问题涉及到几个部分。首先有两种模式：

data = FOREACH source GENERATE
    json#'id_str' AS post_id:chararray,
    json#'user'#'id_str' AS user_id:chararray,
    json#'created_at_str' AS tstamp_str:chararray,
    FLATTEN(json#'entities'#'hashtags') AS hashtag:chararray;

stopwords = FOREACH another_source GENERATE
    token AS stopword_hashtag:chararray;
``` `stopwords` 表包含一些重复项和空值，所以我首先要做的是

stopwords = FILTER stopwords BY stopword_hashtag IS NOT NULL;
stopwords = DISTINCT stopwords;

然后我想过滤掉 `stopwords` 从 `data` 的hashtags，所以我使用filter进行连接，然后将其投影回 `data` :

joined = JOIN data BY hashtag LEFT, stopwords BY stopword_hashtag;
joined = FILTER joined BY stopwords::stopword_hashtag IS NULL;

data = FOREACH joined GENERATE
data::post_id AS post_id:chararray,
data::user_id AS user_id:chararray,
parse_time(data::tstamp_str) AS tstamp:long,
data::hashtag AS hashtag:chararray;
``` parse_time(char array) 是一个 Jython 我编写的udf将datetime字符串转换为unix长时间戳。所有这些之后，我将进行分组+排序：

user_groups = FOREACH (GROUP data BY user_id) GENERATE
    group AS user_id:chararray,
    data.(hashtag, tstamp) AS time_series:bag{tuple:(tag:chararray,tstamp:long)};
user_groups = FOREACH user_groups {
    sorted = ORDER time_series BY tstamp ASC;
    GENERATE user_id, sorted;
}

所有这些都是在一个Pig脚本。我在大数据上运行此进程时遇到性能问题。我知道它适用于小玩具的例子。对于大型示例，其中 data 以及 stopwords 如果时间长，它会占用太多的内存，并且变得非常慢。
据我所知，我正在尽可能早地做过滤，也只做左连接。对性能优化有什么建议吗？
很抱歉更新太晚了。我试过@ran locar方法，效果很不错！
在过滤步骤中，pig中有一个错误：

output = FILTER data_with_count BY from_stopwords_count==0;

报告

java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.Long

我花了一些时间才想出解决办法。我通过显式地将\u stopwords \u count和0转换为long来解决这个问题，即：

data_with_count = FOREACH data_with_count GENERATE
     $0.., (long) from_stopwords_count AS from_stopwords_count:long;
 output = FILTER data_with_count BY from_stopwords_count==0L;

我想我本可以将两者都转换为int，这可能会提高性能，但考虑到我拥有的表的大小，我觉得更安全。
我输入的一些数据：
数据源是gzip格式的1.1tb
stopwords的源代码是400gb的gzip格式，但我只取了其中的一列。
查询运行2.3小时，占用550gb内存和180个vCore。最后成功完成，真的救了我一天！