JavaRDD<字符串>文本文件=sc。文本文件(“hdfs: / /……”);JavaPairRDD<字符串,整数>计数=文本文件。flatMap(年代- >数组。asList(年代。分裂(”“))。迭代器())。mapToPair(词- >新Tuple2< > (词,1))。reduceByKey((一个,b)- >一个+b);计数。saveAsTextFile(“hdfs: / /……”);

列表<整数>l=新ArrayList< > (NUM_SAMPLES);为(int我=0;我<NUM_SAMPLES;我+ +){l。添加(我);}长数=sc。并行化(l)。过滤器(我- >{双x=数学。随机();双y=数学。随机();返回x*x+y*y<1;})。数();系统。出。println(“π”大概是+4.0*数/NUM_SAMPLES);

/ /创建一个DataFrame有一个专栏叫“线”JavaRDD<字符串>文本文件=sc。文本文件(“hdfs: / /……”);JavaRDD<行>rowRDD=文本文件。地图(RowFactory::创建);列表<StructField>字段=数组。asList(数据类型。createStructField(“行”,数据类型。StringType,真正的));StructType模式=数据类型。createStructType(字段);DataFramedf=sqlContext。createDataFrame(rowRDD,模式);DataFrame错误=df。过滤器(上校(“行”)。就像(“% %”错误));/ /计数所有的错误错误。数();/ /计数错误提到MySQL错误。过滤器(上校(“行”)。就像(MySQL“% %”))。数();/ /获取MySQL错误作为一个字符串数组错误。过滤器(上校(“行”)。就像(MySQL“% %”))。收集();

/ /创建一个DataFrame基于一个名为“人”的表/ /存储在一个MySQL数据库。字符串url=" jdbc: mysql: / / yourIP: yourPort /测试?用户= yourUsername;密码= yourPassword”;DataFramedf=sqlContext。读()。格式(“jdbc”)。选项(“url”,url)。选项(“数据表”,“人”)。负载();/ /看起来这个DataFrame的模式。df。printSchema();/ /计数人按年龄DataFramecountsByAge=df。groupBy(“年龄”)。数();countsByAge。显示();/ /保存countsByAge S3的JSON格式。countsByAge。写()。格式(“json”)。保存(“s3a: / /……”);

/ /每一个这个DataFrame包含标签和记录/ /功能由一个向量表示。StructType模式=新StructType(新StructField[]{新StructField(“标签”,数据类型。倍增式,假,元数据。空()),新StructField(“特征”,新VectorUDT(),假,元数据。空()),});DataFramedf=jsql。createDataFrame(数据,模式);/ /设置参数的算法。/ /在这里,我们限制的迭代数到10。LogisticRegressionlr=新LogisticRegression()。setMaxIter(10);/ /符合模型的数据。LogisticRegressionModel模型=lr。适合(df);/ /检查模型:功能权重。向量权重=模型。权重();/ /给定一个数据集,预测每个点的标签,并显示结果。模型。变换(df)。显示();