Apache Spark SQL to narzędzie do „przetwarzania SQL i danych strukturalnych” w Spark, szybkim i uniwersalnym systemie przetwarzania klastrów. Może być używany do pobierania danych z Hive, Parquet itp. Oraz do uruchamiania zapytań SQL na istniejących RDD i Datasets.
Zastanawiam się tylko, jaka jest różnica między RDDi DataFrame (Spark 2.0.0 DataFrame to zwykły alias typu Dataset[Row]) w Apache Spark? Czy potrafisz konwertować jeden na
Załóżmy, że robię coś takiego: val df = sqlContext.load("com.databricks.spark.csv", Map("path" -> "cars.csv", "header" -> "true")) df.printSchema() root |-- year: string (nullable = true) |-- make: string (nullable = true) |-- model: string (nullable = true) |-- comment: string...
Próbowałem, df.orderBy("col1").show(10)ale posortowano w kolejności rosnącej. df.sort("col1").show(10)sortuje również w porządku malejącym. Spojrzałem na stackoverflow i wszystkie odpowiedzi, które znalazłem, były nieaktualne lub odnosiły się do RDD . Chciałbym użyć natywnej ramki danych w...
Mam DataFrame wygenerowaną w następujący sposób: df.groupBy($"Hour", $"Category") .agg(sum($"value") as "TotalValue") .sort($"Hour".asc, $"TotalValue".desc)) Wyniki wyglądają następująco: +----+--------+----------+ |Hour|Category|TotalValue| +----+--------+----------+ | 0| cat26| 30.9| | 0|...
Chciałbym przeczytać plik CSV w Spark i przekonwertować go na DataFrame i zapisać w HDFS z df.registerTempTable("table_name") Próbowałem: scala> val df = sqlContext.load("hdfs:///csv/file/dir/file.csv") Błąd, który otrzymałem: java.lang.RuntimeException: hdfs:///csv/file/dir/file.csv is not...
Jak mogę przekonwertować RDD ( org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]) na Dataframe org.apache.spark.sql.DataFrame. Przekonwertowałem ramkę danych na rdd za pomocą .rdd. Po przetworzeniu chcę go z powrotem w ramce danych. W jaki sposób mogę to zrobić
Chcę dodać kolumnę DataFramez dowolną wartością (taką samą dla każdego wiersza). Pojawia się błąd, gdy używam withColumnw następujący sposób: dt.withColumn('new_column', 10).head(5) --------------------------------------------------------------------------- AttributeError Traceback (most recent...
Mam Spark DataFrame (przy użyciu PySpark 1.5.1) i chciałbym dodać nową kolumnę. Próbowałem następujących bez powodzenia: type(randomed_hours) # => list # Create in Python and transform to RDD new_col = pd.DataFrame(randomed_hours, columns=['new_col']) spark_new_col =...
Zacząłem używać Spark SQL i DataFrames w Spark 1.4.0. Chcę zdefiniować niestandardowy partycjoner w DataFrames w Scali, ale nie widzę, jak to zrobić. Jedna z tabel danych, z którymi pracuję, zawiera listę transakcji według konta, jak w poniższym przykładzie. Account Date Type Amount 1001...
Jak połączyć dwie kolumny w ramce danych Apache Spark? Czy w Spark SQL jest jakaś funkcja, której możemy
Próbuję filtrować ramkę danych PySpark, która ma Nonejako wartość wiersza: df.select('dt_mvmt').distinct().collect() [Row(dt_mvmt=u'2016-03-27'), Row(dt_mvmt=u'2016-03-28'), Row(dt_mvmt=u'2016-03-29'), Row(dt_mvmt=None), Row(dt_mvmt=u'2016-03-30'), Row(dt_mvmt=u'2016-03-31')] i mogę...
Mam ramkę danych z kolumną jako ciągiem. Chciałem zmienić typ kolumny na Double type w PySpark. Oto droga, którą zrobiłem: toDoublefunc = UserDefinedFunction(lambda x: x,DoubleType()) changedTypedf = joindf.withColumn("label",toDoublefunc(joindf['show'])) Chciałem tylko wiedzieć, czy to...
W tej chwili muszę użyć, df.count > 0aby sprawdzić, czy DataFramejest pusty, czy nie. Ale jest to trochę nieefektywne. Czy jest lepszy sposób, aby to zrobić? Dzięki. PS: Chcę sprawdzić, czy jest pusty, aby zapisać tylko DataFramewtedy, gdy nie jest pusty
Chcę tworzyć dalej DataFramez określonym schematem w Scali. Próbowałem użyć odczytu JSON (mam na myśli czytanie pustego pliku), ale nie sądzę, że to najlepsza praktyka.
Próbuję przekonwertować wszystkie nagłówki / nazwy kolumn DataFramew Spark-Scala. na razie wymyślam następujący kod, który zastępuje tylko jedną nazwę kolumny. for( i <- 0 to origCols.length - 1) { df.withColumnRenamed( df.columns(i), df.columns(i).toLowerCase ); }...
Używam pyspark (Python 2.7.9 / Spark 1.3.1) i mam Dataframe GroupObject, którą muszę filtrować i sortować w kolejności malejącej. Próbuję to osiągnąć za pomocą tego fragmentu kodu. group_by_dataframe.count().filter("`count` >= 10").sort('count', ascending=False) Ale generuje następujący...
Chcę przekonwertować kolumnę z ciągiem ramki danych na listę. To, co mogę znaleźć w DataframeAPI, to RDD, więc najpierw próbowałem przekonwertować go z powrotem na RDD, a następnie zastosować toArrayfunkcję do RDD. W takim przypadku długość i SQL działają dobrze. Jednak wynik, który otrzymałem z...
Czy można zapisać DataFramew Spark bezpośrednio w Hive? Próbowałem przekonwertować DataFramedo, Rdda następnie zapisać jako plik tekstowy, a następnie załadować w gałęzi. Ale zastanawiam się, czy mogę bezpośrednio zapisać dataframedo ula
Tak więc, jak wiem, w Spark Dataframe, dla wielu kolumn może mieć taką samą nazwę, jak pokazano poniżej migawka dataframe: [ Row(a=107831, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0}), a=107831, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0})), Row(a=107831,...
>>> a DataFrame[id: bigint, julian_date: string, user_id: bigint] >>> b DataFrame[id: bigint, quan_created_money: decimal(10,0), quan_created_cnt: bigint] >>> a.join(b, a.id==b.id, 'outer') DataFrame[id: bigint, julian_date: string, user_id: bigint, id: bigint,...