经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置:技术经验 » 数据库/运维 » Spark » 查看文章
Spark 用户自定义函数 Java 示例
来源:cnblogs  作者:hapjin  时间:2018/12/3 9:52:53  对本文有异议

Spark UDF Java 示例

这篇文章中提到了用Spark做用户昵称文本聚类分析,聚类需要选定K个中心点,然后迭代计算其他样本点到中心点的距离。由于中文文字分词之后(n-gram)再加上昵称允许各个特殊字符(数字、字母、各种符号……),如果直接在原来的文本数据上进行聚类,由于文本的“多样性”,聚类效果并不一定好。因此准确对昵称先进行一个预分类的过程,这里的分类不是机器学习里面的分类算法(逻辑回归、线性回归),而是根据昵称文本的特征进行分类:给定一个文本昵称字符串,分类方法逐个地将每个字符转换成定义好的模式:

即将所有汉字替换成H、所有大写字母替换成U、小写字母替换成L、数字替换成N,其他各种符号替换成O,然后系统将每一个字符串表示成能标识其组成的字符串模式。
例如用户名:“你好abc123”会被表示成HHLLLNNN,而“你好aaa456”也会被标识成HHLLLNNN。
这样相同的字符串模式的字符将获得相同的表示。

而这里采用Spark 用户自定义函数来实现这种转换。然后再在每个预分类下,进行聚类。

Spark 用户自定义函数介绍

在Java里面通过实现接口UDF(一共定义了22个吧,根据不同参数个数进行选择)来定义一个Spark UDF,简单一点的UDF可以使用Lambda表达式。具体介绍可参考官方文档。如下的NickFormatterUDF接收一个字符串作为输入,将该字符串转换成 由 HLUNWO 组成的字符串模式。

  1. import org.apache.spark.sql.api.java.UDF1;
  2. /**
  3. * @author psj
  4. * @date 2018/11/16
  5. */
  6. public class NickFormatterUDF implements UDF1<String, String> {
  7. @Override
  8. public String call(String nick) throws Exception {
  9. StringBuilder pattern = new StringBuilder();
  10. for (int i = 0; i < nick.length(); i++) {
  11. char ch = nick.charAt(i);
  12. if (ParseChar.isChinese(ch)) {
  13. pattern.append('H');
  14. } else if (ParseChar.isLowerCase(ch)) {
  15. pattern.append('L');
  16. } else if (ParseChar.isUpperCase(ch)) {
  17. pattern.append('U');
  18. } else if (ParseChar.isNumber(ch)) {
  19. pattern.append('N');
  20. } else if (ParseChar.isWhiteSpace(ch)) {
  21. pattern.append('W');
  22. }else{
  23. pattern.append('O');
  24. }
  25. }//END FOR
  26. return pattern.toString();
  27. }
  28. }

ParseChar.java就是一个简单地判断某个字符是中文字符、还是数字、还是大写字母、还是小写字母的工具类。

  1. public class ParseChar {
  2. public static boolean isChinese(char ch) {
  3. //获取此字符的UniCodeBlock
  4. Character.UnicodeBlock ub = Character.UnicodeBlock.of(ch);
  5. // GENERAL_PUNCTUATION 判断中文的“号
  6. // CJK_SYMBOLS_AND_PUNCTUATION 判断中文的。号
  7. // HALFWIDTH_AND_FULLWIDTH_FORMS 判断中文的,号
  8. if (ub == Character.UnicodeBlock.CJK_UNIFIED_IDEOGRAPHS || ub == Character.UnicodeBlock.CJK_COMPATIBILITY_IDEOGRAPHS
  9. || ub == Character.UnicodeBlock.CJK_UNIFIED_IDEOGRAPHS_EXTENSION_A || ub == Character.UnicodeBlock.CJK_UNIFIED_IDEOGRAPHS_EXTENSION_B
  10. || ub == Character.UnicodeBlock.CJK_SYMBOLS_AND_PUNCTUATION || ub == Character.UnicodeBlock.HALFWIDTH_AND_FULLWIDTH_FORMS
  11. || ub == Character.UnicodeBlock.GENERAL_PUNCTUATION) {
  12. // System.out.println(ch + " 是中文");
  13. return true;
  14. }
  15. return false;
  16. }
  17. public static boolean isNumber(char ch) {
  18. return ch >= '0' && ch <= '9';
  19. }
  20. public static boolean isLowerCase(char ch) {
  21. return ch >= 'a' && ch <= 'z';
  22. }
  23. public static boolean isUpperCase(char ch) {
  24. return Character.isUpperCase(ch);
  25. }
  26. public static boolean isWhiteSpace(char ch) {
  27. return Character.isWhitespace(ch);
  28. }
  29. }

写完了自定义函数,接下来看看在Spark中如何调用自己定义的函数。在这里我碰到了很多奇怪的问题。我们的样本数据如下:

{"created":1542020126816,"nick":"a357410","uid":123456}

{"created":1542020138522,"nick":"alichao","signature":"┌?┐?┊雨┊?┊蒙┊?┊蒙┊?└?┘","uid":123456}

{"created":1542020127633,"details":"走过了之后才明白,往事是用来回忆的,幸福是用来感受的,伤痛是用来成长的。。。。","nick":"菲儿","signature":"游戏有你更精彩","uid":123456}

可以看出,样本数据中即有昵称字符、又有签名字段、还有created 字段……而我们只针对昵称字段进行预分类。

首先将样本数据nick_class.json上传到HDFS:

./bin/hdfs dfs -put ~/data_spark/nick_class.json /user/xxx/

然后程序中加载数据:

  1. Dataset<Row> dataset = spark.read().format("json").option("header", "false")
  2. .load("hdfs://localhost:9000/user/xxx/nick_class.json");
  3. dataset.show(10);

对样本中昵称为空的字段进行过滤,并只选取昵称字段应用到Spark UDF上:

  1. Dataset<Row> nickDataset = dataset.filter(col("nick").isNotNull()).select(col("nick"));
  2. nickDataset.show();
  3. nickDataset.printSchema();

先来一个简单一点的Spark中内置的UDF函数:小写字母转换成大写,哈哈。

  1. UserDefinedFunction mode = udf(
  2. (String nick) -> nick.toUpperCase(), DataTypes.StringType);
  3. Dataset upperNickDataFrame = nickDataset.select(mode.apply(col("nick")));
  4. System.out.println(upperNickDataFrame.count());
  5. upperNickDataFrame.show();

对昵称字符串进行分类的用户自定义函数NickFormatterUDF.java。创建对象注册到SparkSession中即可,创建临时视图就是方便后面可通过SQL形式对nickDataFrame进行调用。

  1. UDF1<String, String> nickPreClassificationUDF = new NickFormatterUDF();
  2. spark.udf().register("nick_classifier", nickPreClassificationUDF, DataTypes.StringType);
  3. nickDataset.createOrReplaceTempView("nickDataFrame");

通过functions.callUDF调用UDF:

  1. Dataset nickClassifyDF = nickDataset.select(functions.callUDF("nick_classifier", col("nick")));
  2. System.out.println(nickClassifyDF.count());
  3. nickClassifyDF.show();

Spark作业提交运行

这篇文章中介绍搭建Spark远程调试开发环境,本以为能够在远程调试环境中运行,但每次执行到 upperNickDataFrame.show();或者 nickClassifyDF.show();就抛出异常:

  1. java.lang.ClassCastException: cannot assign instance of scala.collection.immutable.List$SerializationProxy to field org.apache.spark.rdd.RDD.org$apache$spark$rdd$RDD$$dependencies_ of type scala.collection.Seq in instance of org.apache.spark.rdd.MapPartitionsRDD
  2. at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues

一直以为是自定义函数的bug,找了好久没有找到原因,后来在SPARK-18075发现:原来是Spark提交作业的方式有问题。

在自己的Intellij 开发环境下以debug调试运行Spark应用程序固然方便,但这不符合官方推荐的以打成jar包的方式运行Spark作业这种方式。

It is very convenient to write Spark code in an IDE as part of a larger application framework and test it in development by simply running the main function, instead of packaging it into a jar for every single change and submitting this jar to a cluster. Often you have to run it on a remote cluster even for development, especially when handling large quantities of data.

于是:mvn package 将这种工程打成nick_classifier.jar包,上传到服务器上。以命令:./spark-2.3.1-bin-hadoop2.7/bin/spark-submit --class net.hapjin.spark.nick.SparkNickPreClassification nick_classifier.jar

结果还是报同样的错误,或者连接拒绝错误。出现这个问题,主要是环境配置不一致的问题:

  • 程序代码里面创建SparkSession时,需要指定Spark Master地址,这个地址是填 spark://ip:port,还是填spark://master_name:port,还是填spark://localhost:port这个要视集群配置而定了。
  • 第二个是:/etc/hosts里面配置的主机名到ip地址的映射
  • 第三个是conf/spark-env.sh里面的参数:SPARK_LOCAL_IP的设置。

记录一下我在实验环境下运行的结果:

spark-2.3.1-bin-hadoop2.7、hadoop-2.7.7、按hadoop官网的Standalone Operation方式配置启动HDFS。

./spark-2.3.1-bin-hadoop2.7/bin/spark-submit --class net.hapjin.spark.nick.SparkNickPreClassification nick_classifier.jar提交运行。

源码如下:

  1. package net.hapjin.spark.nick;
  2. import org.apache.spark.sql.*;
  3. import org.apache.spark.sql.api.java.UDF1;
  4. import org.apache.spark.sql.expressions.UserDefinedFunction;
  5. import org.apache.spark.sql.types.DataTypes;
  6. import static org.apache.spark.sql.functions.col;
  7. import static org.apache.spark.sql.functions.udf;
  8. /**
  9. * @author psj
  10. * @date 2018/11/16
  11. */
  12. public class SparkNickPreClassification {
  13. public static void main(String[] args) {
  14. SparkSession spark = SparkSession.builder().appName("nick_classification")
  15. .master("local[*]")
  16. .getOrCreate();
  17. Dataset<Row> dataset = spark.read().format("json").option("header", "false")
  18. .load("hdfs://localhost:9000/user/xxx/nick_class.json");
  19. dataset.show(10);
  20. Dataset<Row> nickDataset = dataset.filter(col("nick").isNotNull()).select(col("nick"), col("uid"));
  21. nickDataset.show();
  22. nickDataset.printSchema();
  23. //https://stackoverflow.com/questions/28186607/java-lang-classcastexception-using-lambda-expressions-in-spark-job-on-remote-ser
  24. // spark.udf().register("udfUpperCase", (String string) -> string.toUpperCase(), DataTypes.StringType);
  25. // Dataset<Row> df = nickDataset.withColumn("upper", callUDF("udfUpperCase", nickDataset.col("nick")));
  26. // System.out.println(df.count());
  27. // df.show();
  28. //https://issues.apache.org/jira/browse/SPARK-18075
  29. //没有按标准来运行spark app jar
  30. UserDefinedFunction mode = udf(
  31. (String nick) -> nick.toUpperCase(), DataTypes.StringType);
  32. Dataset upperNickDataFrame = nickDataset.select(mode.apply(col("nick")));
  33. System.out.println(upperNickDataFrame.count());
  34. upperNickDataFrame.show();
  35. UDF1<String, String> nickPreClassificationUDF = new NickFormatterUDF();
  36. spark.udf().register("nick_classifier", nickPreClassificationUDF, DataTypes.StringType);
  37. nickDataset.createOrReplaceTempView("nickDataFrame");
  38. // spark.sql("select uid, nick_classifier(nick) from nickDataFrame").show();
  39. Dataset nickClassifyDF = nickDataset.select(functions.callUDF("nick_classifier", col("nick")), col("nick"), col("uid"));
  40. System.out.println(nickClassifyDF.count());
  41. //https://stackoverflow.com/questions/39953245/how-to-fix-java-lang-classcastexception-cannot-assign-instance-of-scala-collect
  42. nickClassifyDF.show();
  43. }
  44. }

最终运行出来的结果:可以看出已经成功地将昵称转换成 自定义的 字符模式。左边列就是每个昵称的模式,右边列是实际的昵称。比如第一行:UU 代表两个大写的英文字符,而左边的nick是"JX"(意味着将所有 两个大写字母 的昵称 转换成类别 UU 了)

后面会将 这些自定义的字符模式 归为几个类别,然后在每个类别上进行聚类分析。

参考资料:

[基于层次聚类的虚假用户检测]

Spark Java API 计算 Levenshtein 距离

Spark Java API 之 CountVectorizer

spark JAVA 开发环境搭建及远程调试

 友情链接:直通硅谷  点职佳  北美留学生论坛

本站QQ群:前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

W3xue 的所有内容仅供测试,对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接(目前9元/月)请联系QQ:27243702 沸活量
皖ICP备17017327号-2 皖公网安备34020702000426号