使用 Spark 以 CSV 格式提取部分字符串

Question

在这里激发新手，希望你们能给我一些帮助。谢谢！

我正在尝试从 CSV 文件中提取 URL，该 URL 位于第 16 列。问题是 URL 是以奇怪的格式编写的，您可以从下面的代码的打印输出中看到。获取正确格式的 URL 的最佳方法是什么？

case class log(time_stamp: String, url: String )
val logText = sc.textFile("hdfs://...").map(s => s.split(",")).map( s => log(s(0).replaceAll(""", ""),s(15).replaceAll(""", ""))).toDF()

logText.registerTempTable("log")
val results = sqlContext.sql("SELECT * FROM log")
results.map(s => "URL: " + s(1)).collect().foreach(println)
URL: /XXX/YYY/ZZZ/http/www.domain.com/xyz/xyz
URL: /XX/YYY/ZZZ/https/sub.domain.com/xyz/xyz/xyz/xyz
URL: /XXX/YYY/ZZZ/http/www.domain.com/
URL: /VV/XXXX/YYY/ZZZ/https/sub.domain.com/xyz/xyz/xyz

Answer 1

你可以试试regexp_replace：

import org.apache.spark.sql.functions.regexp_replace
val df = sc.parallelize(Seq(
  (1L, "/XXX/YYY/ZZZ/http/www.domain.com/xyz/xyz"),
  (2L, "/XXX/YYY/ZZZ/https/sub.domain.com/xyz/xyz/xyz/xyz")
)).toDF("id", "url")
df
  .select(regexp_replace($"url", "^(/\w+){3}/(https?)/", "$2://").alias("url"))
  .show(2, false)
// +--------------------------------------+
// |url                                   |
// +--------------------------------------+
// |http://www.domain.com/xyz/xyz         |
// |https://sub.domain.com/xyz/xyz/xyz/xyz|
// +--------------------------------------+

在 Spark 1.4 中，你可以尝试 Hive UDF：

df.selectExpr("""regexp_replace(url, '^(/w+){3}/(https?)/','$2://') AS url""")

如果http(s)之前的部分数可以变化，您可以通过将{3}替换为*或范围来调整正则表达式。

Answer 2

问题归结为解析长字符串并提取域名。只要您没有任何随机字符串（XXX，YYYYY等）是"http"和"https"，此解决方案就可以工作：

def getUrl(data: String): Option[String] = {
   val slidingPairs = data.split("/").sliding(2)
   slidingPairs.flatMap{ 
       case Array(x,y) => 
          if(x=="http" || x == "https") Some(y) else None
    }.toList.headOption
}

以下是 REPL 中的一些示例：

scala> getUrl("/XXX/YYY/ZZZ/http/www.domain.com/xyz/xyz")
res8: Option[String] = Some(www.domain.com)
scala> getUrl("/XXX/YYY/ZZZ/https/sub.domain.com/xyz/xyz/xyz/xyz")
resX: Option[String] = Some(sub.domain.com)
scala> getUrl("/XXX/YYY/ZZZ/a/asdsd/asdase/123123213/https/sub.domain.com/xyz/xyz/xyz/xyz")
resX: Option[String] = Some(sub.domain.com)