小编add*_*ing的帖子

我正在使用带有Anaconda发行版的PyCharm IDE .当我运行时:Tools > Python Console...PyCharm使用ipython控制台,它是Anaconda发行版的一部分.

但它使用默认配置文件.我已经尝试--profile=myProfileName在环境变量和解释器选项中添加选项Settings > Build, Execution, Deployment > Console > Python Console 但它继续使用默认配置文件.

我的问题是如何ipython在PyCharm中设置不同的配置文件？

我有大卡桑德拉桌子。我只想从Cassandra加载50行。以下代码

val ds = sparkSession.read
      .format("org.apache.spark.sql.cassandra")
      .options(Map("table" -> s"$Aggregates", "keyspace" -> s"$KeySpace"))
      .load()
      .where(col("aggregate_type") === "DAY")
      .where(col("start_time") <= "2018-03-28")
      .limit(50).collect()

以下代码将两个谓词从where方法中推入，但不限制一个。是否获取了整个数据（100万条记录）？如果不是，为什么此代码的运行时间与没有代码的代码limit(50)大致相同。

给出以下代码：

import java.sql.Date
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

object SortQuestion extends App{

  val spark = SparkSession.builder().appName("local").master("local[*]").getOrCreate()
  import spark.implicits._
  case class ABC(a: Int, b: Int, c: Int)

  val first = Seq(
    ABC(1, 2, 3),
    ABC(1, 3, 4),
    ABC(2, 4, 5),
    ABC(2, 5, 6)
  ).toDF("a", "b", "c")

  val second = Seq(
    (1, 2, (Date.valueOf("2018-01-02"), 30)),
    (1, 3, (Date.valueOf("2018-01-01"), 20)),
    (2, 4, (Date.valueOf("2018-01-02"), 50)),
    (2, 5, (Date.valueOf("2018-01-01"), 60))
  ).toDF("a", "b", "c")

  first.join(second.withColumnRenamed("c", "c2"), Seq("a", "b")).groupBy("a").agg(sort_array(collect_list("c2")))
    .show(false)

}

Spark产生以下结果：

import java.sql.Date
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._ …

我想根据基于最低 k-means 分数的 'k' 参数选择 k-means 模型。

我可以手动找到“k”参数的最佳值，写一些类似的东西

  def clusteringScore0(data: DataFrame, k: Int): Double = {
    val assembler = new VectorAssembler().
      setInputCols(data.columns.filter(_ != "label")).
      setOutputCol("featureVector")
    val kmeans = new KMeans().
      setSeed(Random.nextLong()).
      setK(k).
      setPredictionCol("cluster").
      setFeaturesCol("featureVector")
    val pipeline = new Pipeline().setStages(Array(assembler, kmeans))
    val kmeansModel = pipeline.fit(data).stages.last.asInstanceOf[KMeansModel]
    kmeansModel.computeCost(assembler.transform(data)) / data.count()   }

  (20 to 100 by 20).map(k => (k, clusteringScore0(numericOnly, k))).
    foreach(println)

我应该使用 CrossValitor API 吗？

像这样的东西：

val paramGrid = new ParamGridBuilder().addGrid(kmeansModel.k, 20 to 100 by 20).build()
val cv = new CrossValidator().setEstimator(pipeline).setEvaluator(new KMeansEvaluator()).setEstimatorParamMaps(paramGrid).setNumFolds(3)

有用于回归和分类的评估器，但没有用于聚类的评估器。 …

给定 Cassandra 表：

CREATE TABLE data_storage.stack_overflow_test_table (
    id int,
    text_id text,
    clustering date,
    some_other text,
    PRIMARY KEY (( id, text_id ), clustering)
)

以下查询是有效查询：

select * from data_storage.test_table_filtering where id=4 and text_id='2';

因为我包含了从分区键到查询的所有列。

考虑以下代码：

val ds = session.
  read
  .format("org.apache.spark.sql.cassandra")
  .options(Map("table" -> "stack_overflow_test_table", "keyspace" -> "data_storage"))
  .load()
  .where(col("id") === 4 &&
  col("text_id") === "2").show(10)

由于 spark-cassandra 连接器将谓词推送到 Cassandra，我希望 Spark 将发送 Cassandra 的查询类似于

SELECT "id", "text_id", "clustering", "some_other" FROM "data_storage"."stack_overflow_test_table" WHERE "id" = ? AND "text_id" = ? 

但是，我可以在日志中看到

18/04/09 15:38:09 …

我有两个DataFrames推荐和电影.建议中的列rec1-rec3表示电影数据帧中的电影ID.

val recommendations: DataFrame = List(
        (0, 1, 2, 3),
        (1, 2, 3, 4),
        (2, 1, 3, 4)).toDF("id", "rec1", "rec2", "rec3")

val movies = List(
        (1, "the Lord of the Rings"),
        (2, "Star Wars"),
        (3, "Star Trek"),
        (4, "Pulp Fiction")).toDF("id", "name")

我想要的是:

+---+------------------------+------------+------------+
| id|                    rec1|        rec2|        rec3|
+---+------------------------+------------+------------+
|  0|   the Lord of the Rings|   Star Wars|   Star Trek|
|  1|               Star Wars|   Star Trek|Pulp Fiction|
|  2|   the Lord of the Rings|   Star Trek|   Star Trek|
+---+------------------------+------------+------------+

我想获取 DataFrame 的所有列。如果 DataFrame 具有平面结构（没有嵌套的 StructTypes），则会df.columns产生正确的结果。我也想返回所有嵌套的列名，例如

给定的

val schema = StructType(
  StructField("name", StringType) ::
  StructField("nameSecond", StringType) ::
  StructField("nameDouble", StringType) ::
  StructField("someStruct", StructType(
    StructField("insideS", StringType)::
    StructField("insideD", DoubleType)::
    Nil
  )) ::
  Nil
)
val rdd = spark.sparkContext.emptyRDD[Row]
val df = spark.createDataFrame(rdd, schema)

我想得到

Seq("name", "nameSecond", "nameDouble", "someStruct", "insideS", "insideD")

我在工作中使用 pyspark。在这篇文章https://unraveldata.com/to-cache-or-not-to-cache/中，它说缓存不是一个动作。然而，当我在 RDD 上运行缓存函数时，需要花费很多时间。Spark UI 显示有一些名为 的激活作业cache at NativeMethodAccessorImpl.java:0。那么缓存是一个动作吗？