Циклевка паркета что это: что это такое, как отциклевать без пыли и с покрытием лака старое напольное покрытие без выноса мебели

Краткое введение в PySpark. PySpark — отличный язык для… | Бен Вебер Источник: Wikimedia (Венский технический музей)

Учебник по PySpark для науки о данных

PySpark — отличный язык для проведения аналитического анализа данных в масштабе, построения конвейеров машинного обучения и создания ETL для платформы данных. Если вы уже знакомы с Python и такими библиотеками, как Pandas, то PySpark — это отличный язык для изучения с целью создания более масштабируемых анализов и конвейеров.Цель этого поста — показать, как начать работу с PySpark и выполнить общие задачи.

Мы будем использовать блоки данных для среды Spark, а набор данных НХЛ от Kaggle в качестве источника данных для анализа. В этом посте показано, как читать и записывать данные в кадры данных Spark, создавать преобразования и объединения этих кадров, визуализировать результаты и выполнять линейную регрессию. Я также покажу, как смешивать обычный код Python с PySpark в масштабируемой форме, используя пользовательские функции Pandas. Для простоты мы сосредоточимся на пакетной обработке и избежим некоторых сложностей, возникающих при потоковой передаче данных.

Полный блокнот для этого поста доступен на github.

Environment

Существует несколько различных вариантов запуска и запуска с Spark:

• Self Hosted: Вы можете самостоятельно настроить кластер, используя «голые железные» машины или виртуальные машины. Apache Ambari — полезный проект для этой опции, но я не рекомендую подход для быстрого запуска и запуска.
• Облачные провайдеры: Большинство облачных провайдеров предлагают кластеры Spark: в AWS есть EMR, а в GCP — DataProc.В прошлом я писал о DataProc, и вы можете быстрее перейти в интерактивную среду, чем самостоятельно.
• Решения для поставщиков: Компании, в том числе Databricks и Cloudera, предоставляют решения Spark, облегчающие запуск и эксплуатацию Spark.

Используемое решение зависит от безопасности, стоимости и существующей инфраструктуры. Если вы пытаетесь настроить среду для обучения, я бы предложил использовать Databricks Community Edition.

Создание кластера PySpark в Databricks Community Edition.

В этой среде легко начать работу с кластером Spark и средой ноутбука. Для этого урока я создал кластер со средой исполнения Spark 2.4 и Python 3. Для запуска кода в этом посте вам понадобится как минимум Spark версии 2.3 для функциональных возможностей UDF Pandas.

Spark Dataframes

Ключевым типом данных, используемым в PySpark, является Spark dataframe. Этот объект можно рассматривать как таблицу, распределенную по кластеру, и его функциональность аналогична фреймам данных в R и Pandas.Если вы хотите выполнять распределенные вычисления с использованием PySpark, то вам нужно будет выполнять операции с кадрами данных Spark, а не с другими типами данных Python.

Также возможно использовать кадры данных Pandas при использовании Spark, вызывая toPandas () на кадре данных Spark, который возвращает объект Pandas. Тем не менее, эту функцию, как правило, следует избегать, за исключением случаев работы с небольшими кадрами данных, поскольку она вытягивает весь объект в память на одном узле.

Одним из ключевых отличий между фреймами данных Pandas и Spark является стремление к ленивому выполнению.В PySpark операции откладываются до тех пор, пока в конвейере фактически не потребуется результат. Например, вы можете указать операции для загрузки набора данных из S3 и применения ряда преобразований к фрейму данных, но эти операции не будут применены немедленно. Вместо этого записывается график преобразований, и как только данные действительно необходимы, например, при записи результатов обратно в S3, преобразования применяются как одна конвейерная операция. Этот подход используется, чтобы избежать вытягивания полного фрейма данных в память и обеспечивает более эффективную обработку на кластере машин.С информационными кадрами Pandas все вытягивается в память, и каждая операция Pandas немедленно применяется.

Как правило, рекомендуется избегать активных операций в Spark, если это возможно, поскольку это ограничивает объем вашего конвейера, который может быть эффективно распределен.

Чтение данных

Одним из первых шагов при работе со Spark является загрузка набора данных в фрейм данных. После того, как данные загружены в фрейм данных, вы можете применять преобразования, выполнять анализ и моделирование, создавать визуализации и сохранять результаты.В Python вы можете загружать файлы напрямую из локальной файловой системы, используя Pandas:

 импорт pandas как pd 
 pd.read_csv ("dataset.csv") 

В PySpark загрузка файла CSV немного сложнее. В распределенной среде нет локального хранилища, и поэтому для указания пути к файлу необходимо использовать распределенную файловую систему, такую ​​как HDFS, хранилище файлов Databricks (DBFS) или S3.

Как правило, при использовании PySpark я работаю с данными в S3. Многие базы данных предоставляют функцию выгрузки в S3, а также можно использовать консоль AWS для перемещения файлов с локального компьютера на S3.Для этого поста я буду использовать файловую систему Databricks (DBFS), которая предоставляет пути в виде / FileStore . Первый шаг — загрузить файл CSV, который вы хотите обработать.

Загрузка файла в хранилище файлов Databricks.

Следующим шагом является чтение файла CSV в информационный кадр Spark, как показано ниже. Этот фрагмент кода указывает путь к файлу CSV и передает ряд аргументов в функцию read для обработки файла. Последний шаг отображает подмножество загруженного фрейма данных, аналогично df.руководитель () в пандах.

 file_location = "/FileStore/tables/game_skater_stats.csv"df = spark.read.format (" csv "). Option (" inferSchema ", 
 True) .option (" header ", True) .load (file_location) display (df) 

Я предпочитаю использовать формат паркета при работе со Spark, потому что это формат файла, который включает метаданные о типах данных столбца, предлагает сжатие файла и является форматом файла, который хорошо работает с Spark. AVRO — еще один формат, который хорошо работает со Spark.Ниже приведен фрагмент кода, в котором показано, как извлечь кадр данных из предыдущего фрагмента и сохранить его в виде файла паркета в DBFS, а затем перезагрузить кадр данных из сохраненного файла паркета.

 df.write.save ('/ FileStore / parquet / game_skater_stats', 
 format = 'parquet') df = spark.read.load ("/ FileStore / parquet / game_skater_stats") 
 дисплей (df) 

Результат на этом шаге то же самое, но поток выполнения значительно отличается. При чтении файлов CSV в кадры данных Spark выполняет операцию в активном режиме, что означает, что все данные загружаются в память до начала выполнения следующего шага, тогда как при чтении файлов в формате паркета используется ленивый подход.Как правило, вы хотите избежать активных операций при работе со Spark, и если мне нужно обработать большие CSV-файлы, я сначала преобразую набор данных в формат паркета, прежде чем выполнять остальную часть конвейера.

Часто вам нужно обрабатывать большое количество файлов, например, сотни файлов паркетных файлов, расположенных по определенному пути или в каталоге в DBFS. С помощью Spark вы можете включить подстановочный знак в путь для обработки коллекции файлов. Например, вы можете загрузить пакет паркетных файлов из S3 следующим образом:

 df = spark.читать .load ("s3a: //my_bucket/game_skater_stats/*.parquet") 

Этот подход полезен, если у вас есть отдельный файл паркета в день, или если в вашем конвейере есть предварительный шаг, который выводит сотни файлов паркета ,

Если вы хотите читать данные из базы данных, такой как Redshift, рекомендуется сначала выгружать данные в S3, а затем обрабатывать их в Spark. В Redshift команду unload можно использовать для экспорта данных в S3 для обработки:

 выгрузить ('select * from data_to_process') 
 в 's3: // my_bucket / game_data' 
 iam_role 'arn: aws: iam :: 123 : роль / RedshiftExport '; 

Существуют также библиотеки для баз данных, такие как spark-redshift, которые облегчают выполнение этого процесса.

Запись данных

Аналогично чтению данных в Spark, при использовании PySpark не рекомендуется записывать данные в локальное хранилище. Вместо этого вы должны использовать распределенную файловую систему, такую ​​как S3 или HDFS. Если вы собираетесь обрабатывать результаты с помощью Spark, то паркет — это хороший формат для сохранения фреймов данных. Приведенный ниже фрагмент демонстрирует, как сохранить фрейм данных в DBFS и S3 как паркет.

  # DBFS (паркет) 
  df.write.save ('/ FileStore / parquet / game_stats', format = 'parquet')  # S3 (паркет) 
  df.write.parquet ("s3a: // my_bucket / game_stats", mode = "overwrite") 

При сохранении кадра данных в формате паркета он часто разбивается на несколько файлов, как показано на рисунке ниже.

Файлы паркета, созданные при сохранении кадра данных в DBFS.

Если вам нужны результаты в CSV-файле, требуется немного другой шаг вывода. Одним из основных отличий в этом подходе является то, что все данные будут перенаправлены на один узел перед выводом в CSV. Этот подход рекомендуется, когда вам нужно сохранить небольшой массив данных и обработать его в системе за пределами Spark.Фрагмент ниже показывает, как сохранить фрейм данных в виде одного файла CSV в DBFS и S3.

  # DBFS (CSV) 
  df.write.save ('/ FileStore / parquet / game_stats.csv', format = 'csv')  # S3 (CSV) 
  df.coalesce (1) .write. формат ("com.databricks.spark.csv") 
 .option ("header", "true") .save ("s3a: //my_bucket/game_sstats.csv") 

Еще одним распространенным выходом для сценариев Spark является NoSQL базы данных, такие как Cassandra, DynamoDB или Couchbase. Это выходит за рамки этого поста, но один из подходов, который я видел в прошлом, — это запись кадра данных в S3, а затем запуск процесса загрузки, который сообщает системе NoSQL загружать данные из указанного пути на S3.

Я также пропустил запись в потоковый источник вывода, такой как Kafka или Kinesis. Эти системы более полезны при использовании Spark Streaming.

Преобразование данных

Множество различных типов операций могут быть выполнены на фреймах данных Spark, во многом подобно широкому спектру операций, которые могут применяться к фреймам данных Pandas. Одним из способов выполнения операций над кадрами данных Spark является Spark SQL, который позволяет запрашивать кадры данных, как если бы они были таблицами.Фрагмент ниже показывает, как найти лучших игроков в наборе данных.

 df.createOrReplaceTempView ("stats") отображать (spark.sql ("" "
 выберите player_id, сумму (1) в качестве игр, сумму (цели) в качестве целей 
 из группы статистики 
 на 1 
 порядок на 3 деск 
 предел 5 
 "" ")) 

В результате получается список идентификаторов игроков, количество появлений в играх и общее количество голов, забитых в этих играх. Если мы хотим показать имена игроков, нам нужно загрузить дополнительный файл, сделать его доступным как временное представление, а затем присоединиться к нему с помощью Spark SQL.

Лучшие игроки рейтинга в наборе данных.

В приведенном выше фрагменте кода я использовал команду display для вывода выборки набора данных, но также можно присвоить результаты другому кадру данных, который можно использовать на последующих этапах в конвейере. Приведенный ниже код показывает, как выполнить эти шаги, когда результаты первого запроса назначаются новому фрейму данных, который затем назначается временному представлению и объединяется с коллекцией имен игроков.

 top_players = spark.sql ("" "
 выберите player_id, sum (1) в качестве игр, sum (goal) в качестве голов 
 из группы 
 статистики на 1 
, порядок на 3 desc 
 limit 5 
" "") top_players.createOrReplaceTempView ("top_players") 
 names.createOrReplaceTempView ("names") display (spark.sql ("" "
 выберите p.player_id, goal, _c1 как First, _c2 как Last 
 из top_players p 
, объедините имена n 
 на p .player_id = n._c0 
 упорядочить по 2 сп. 
 "" ")) 

Результат этого процесса показан ниже, определяя Алексея Овечкина как лучшего игрока в НХЛ, основанного на наборе данных Kaggle.

Выходные данные процесса, соединяющего кадры данных с использованием Spark SQL.

Существуют операции с кадрами данных Spark для общих задач, таких как добавление новых столбцов, удаление столбцов, выполнение объединений и вычисление статистических данных агрегирования и аналитики, но с самого начала эти операции проще выполнить с помощью Spark SQL.Кроме того, проще переносить код с Python на PySpark, если вы уже используете библиотеки, такие как PandaSQL или framequery, для манипулирования кадрами данных Pandas с помощью SQL.

Как и большинство операций над кадрами данных Spark, операции Spark SQL выполняются в режиме отложенного выполнения, что означает, что шаги SQL не будут оцениваться до тех пор, пока не потребуется результат. Spark SQL предоставляет отличный способ копаться в PySpark, без необходимости сначала изучать новую библиотеку для фреймов данных.

Если вы используете блоки данных, вы также можете создавать визуализации прямо в блокноте, без явного использования библиотек визуализации.Например, мы можем построить среднее количество голов за игру, используя приведенный ниже код Spark SQL.

Отображение

 (spark.sql ("" "
 выберите приведение (подстрока (game_id, 1, 4) || '-' 
 || substring (game_id, 5, 2) || '-01' в качестве даты) как месяц) 
, сумма (цели) / количество (отличный от game_id) как goal_per_goal 
 из группы статистики 
 по порядку 1 
 по 1 
 "" ")) 

Первоначальный вывод, отображаемый в записной книжке Databricks, представляет собой таблицу результатов, но мы можем используйте функциональность графика, чтобы преобразовать выходные данные в различные визуализации, такие как гистограмма, показанная ниже.Этот подход не поддерживает каждую визуализацию, которая может понадобиться ученому, но он значительно упрощает проведение аналитического анализа данных в Spark. При необходимости мы можем использовать функцию toPandas () , чтобы создать фрейм данных Pandas на узле драйвера, что означает, что для визуализации результатов может использоваться любая библиотека построения графиков Python. Однако этот подход следует использовать только для небольших фреймов данных, поскольку все данные охотно загружаются в память на узле драйвера.

Среднее количество голов за игру в феврале и марте.

Я также посмотрел на средние голы за удар, для игроков с минимум 5 голами.

Дисплей

 (spark.sql ("" "
 выберите бросок (голы / удары * 50 как int) /50.0 как голы_per_shot 
, сумма (1) как игроки 
 из (
 выберите player_id, сумма (удары) как выстрелы, сумма) (цели) как цели 
 из статистики 
 группа на 1 
 с целями> = 5 
) 
 группа на 1 
 порядок на 1 
 "" ")) 

Результаты этого преобразования приведены в таблице ниже. Большинство игроков с по крайней мере 5 голами выполняют броски примерно от 4% до 12% времени.

голов за удар для игроков в наборе данных Kaggle.

MLlib

Одним из распространенных случаев использования Python для ученых данных является построение прогностических моделей. Хотя scikit-learn хорош при работе с пандами, он не масштабируется до больших наборов данных в распределенной среде (хотя есть способы его параллелизации со Spark). При построении прогностических моделей с использованием PySpark и массивных наборов данных MLlib является предпочтительной библиотекой, поскольку она изначально работает с фреймами данных Spark.Не каждый алгоритм в scikit-learn доступен в MLlib, но существует множество вариантов, охватывающих множество вариантов использования.

Для того, чтобы использовать один из контролируемых алгоритмов в MLib, вам нужно настроить свой фрейм данных с вектором признаков и меткой в ​​качестве скаляра. После подготовки вы можете использовать fit для обучения модели. Приведенный ниже фрагмент демонстрирует, как объединить несколько столбцов в кадре данных в один вектор с вектором , используя VectorAssembler .Получившийся кадр данных мы используем для вызова функции fit , а затем генерируем сводную статистику для модели.

  # MLlib импортирует 
  из pyspark.ml.feature import VectorAssembler 
 из pyspark.ml.regression import LinearRegression  # Создайте векторное представление для объектов 
  ассемблер = VectorAssembler (inputCols = ['shots', 'hit', 'assists', 
 'штрафных минут', 'timeOnIce', 'еда на вынос'], outputCol = "функции") 
 train_df = ассемблер.transform (df)  # Подходит для модели линейной регрессии 
  lr = LinearRegression (featuresCol = 'features', labelCol = 'цели') 
 lr_model = lr.fit (train_df)  # Выходная статистика 
  trainingSummary = lr_model.summary 
 print ("Коэффициенты:" + str (lr_model.coefficients)) 
 print ("RMSE:% f"% trainingSummary.rootMeanSquaredError) 
 print ("R2:% f"% trainingSummary.r2) 

Модель прогнозирует, сколько целей игрок будет забивать в зависимости от количества выстрелов, времени в игре и других факторов.Однако производительность этой модели низкая, что приводит к среднеквадратической ошибке (RMSE) 0,375 и значению R-квадрата 0,125. Коэффициент с наибольшим значением составлял кадров, столбец , но это не обеспечивало достаточного сигнала для точности модели.

При создании конвейера ML с PySpark необходимо учесть ряд дополнительных шагов, включая обучение и тестирование наборов данных, настройку гиперпараметров и хранение моделей. Вышеприведенный фрагмент — просто отправная точка для начала работы с MLlib.

UDF Pandas

Одной из функций Spark, которую я использовал недавно, являются пользовательские функции Pandas (UDF), которые позволяют выполнять распределенные вычисления с кадрами данных Pandas в среде Spark. Общий способ работы этих пользовательских функций заключается в том, что вы сначала разбиваете фрейм данных Spark, используя оператор groupby , и каждый раздел отправляется на рабочий узел и транслируется в фрейм данных Pandas, который передается в UDF. UDF затем возвращает преобразованный кадр данных Pandas, который объединяется со всеми другими разделами и затем переводится обратно в кадр данных Spark.Конечный результат действительно полезен, вы можете использовать библиотеки Python, которые требуют Pandas, но теперь могут масштабироваться до массивных наборов данных, если у вас есть хороший способ разбить ваш фрейм данных. Пользовательские функции Pandas были представлены в Spark 2.3, и я буду говорить о том, как мы используем эту функциональность в Zynga во время Spark Summit 2019.

Подгонка кривых — это обычная задача, которую я выполняю как ученый по обработке данных. Приведенный ниже фрагмент кода показывает, как выполнить подгонку кривой для описания взаимосвязи между количеством выстрелов и попаданий, которые игрок записывает в ходе игры.Фрагмент показывает, как мы можем выполнить эту задачу для одного игрока, вызвав toPandas () для набора данных, отфильтрованного для одного игрока. Результатом этого шага являются два параметра (коэффициенты линейной регрессии), которые пытаются описать взаимосвязь между этими переменными.

  # Пример данных для игрока 
  sample_pd = spark.sql ("" "
 select * from stats 
, где player_id = 8471214 
" "") .toPandas ()  # Импорт библиотек python 
  из scipy.оптимизировать импорт leastsq 
 import numpy as np  # Определить функцию для соответствия 
  def fit (params, x, y): 
 return (y - (params [0] + x * params [1]))  # Fit the кривая и показать результаты  
 result = leastsq (fit, [1, 0], 
 args = (sample_pd.shots, sample_pd.hits)) 
 print (result) 

Если мы хотим рассчитать эту кривую для каждого игрока и Если у вас будет большой набор данных, то вызов toPandas () не будет выполнен из-за нехватки памяти. Мы можем масштабировать эту операцию для всего набора данных, вызвав groupby () на player_id , а затем применив UDF Pandas, показанный ниже.Функция принимает в качестве входных данных фрейм данных Pandas, который описывает статистику игрового процесса для одного игрока, и возвращает итоговый фрейм данных, который включает в себя player_id и соответствующие коэффициенты. Каждый из итоговых фреймов данных Pandas затем объединяется в фрейм данных Spark, который отображается в конце фрагмента кода. Еще одна часть настройки для использования пользовательских функций Pandas — это определение схемы для результирующего кадра данных, где схема описывает формат кадра данных Spark, сгенерированного на этапе применения.

  # Загрузить необходимые библиотеки 
  из pyspark.sql.functions import pandas_udf, PandasUDFType 
 из pyspark.sql.types import * 
 import pandas as pd  # Создать схему для результирующего фрейма данных 
  schema = StructType ([ StructField ('ID', LongType (), True), 
 StructField ('p0', DoubleType (), True), 
 StructField ('p1', DoubleType (), True)])  # Определить UDF, ввод и выходы Pandas DFs 
  @pandas_udf (схема, PandasUDFType.GROUPED_MAP) 
 def analysis_player (sample_pd):  # вернуть пустые параметры в недостаточно данных 
  if (len (sample_pd.shots) <= 1): 
 вернуть pd.DataFrame ({'ID': [sample_pd.player_id [0 ]], 
 'p0': [0], 'p1': [0]}) 
  # Выполнить подбор кривой 
  result = leastsq (fit, [1, 0], args = (sample_pd.shots, 
 sample_pd) .hits)) 
  # Возвращать параметры в виде Pandas DF 
  return pd.DataFrame ({'ID': [sample_pd.player_id [0]], 
 'p0': [result [0] [0]], 'p1': [result [0] [1]]})  # выполнить UDF и показать результаты 
  player_df = df.groupby ('player_id'). apply (analysis_player) 
 дисплей (player_df) 

Результат этого процесса показан ниже. Теперь у нас есть фрейм данных, который суммирует кривую, подходящую для каждого игрока, и мы можем выполнить эту операцию для массивного набора данных. При работе с огромными наборами данных важно выбрать или сгенерировать ключ раздела, чтобы добиться хорошего компромисса между количеством и размером разделов данных.

Выходные данные из UDF Pandas, показывающие кривые для каждого игрока.

Best Practices

Я рассмотрел некоторые общие задачи по использованию PySpark, но также хотел дать несколько советов по упрощению перехода от Python к PySpark.Вот некоторые из лучших рекомендаций, которые я собрал, основываясь на моем опыте переноса нескольких проектов между этими средами:

• Избегайте словарей, используйте кадры данных: с использованием типов данных Python, таких как словари, означает, что код может быть не исполняемым в распределенный режим. Вместо использования ключей для индексации значений в словаре рассмотрите возможность добавления еще одного столбца в фрейм данных, который можно использовать в качестве фильтра.
• Используйте toPandas экономно: Вызов toPandas () приведет к загрузке всех данных в память на узле драйвера и предотвратит выполнение операций в распределенном режиме.Хорошо использовать эту функцию, когда данные уже агрегированы, и вы хотите использовать знакомые инструменты построения графиков Python, но ее не следует использовать для больших фреймов данных.
• Избегайте циклов: Если это возможно, предпочтительно переписать логику циклов с использованием шаблона groupby-apply для поддержки выполнения распараллеленного кода. Я заметил, что сосредоточение внимания на использовании этого шаблона в Python также привело к очистке кода, который легче перевести на PySpark.
• Попробуйте свести к минимуму активные операции: Чтобы ваш конвейер был настолько масштабируемым, насколько это возможно, полезно избегать активных операций, которые помещают в память полные кадры данных.Я заметил, что чтение в CSV — это сложная операция, и моя задача заключается в том, чтобы сохранить фрейм данных в виде паркета, а затем перезагрузить его из паркета для создания более масштабируемых конвейеров.
• Используйте framequery / pandasql, чтобы упростить портирование: Если вы работаете с чужим кодом Python, может быть сложно понять, чего достигают некоторые операции Pandas. Если вы планируете переносить код из Python в PySpark, то использование библиотеки SQL для Pandas может упростить этот перевод.

Я обнаружил, что написание кода в PySpark также улучшилось благодаря навыкам программирования на Python.

Заключение

PySpark — отличный язык для изучения данных, поскольку он позволяет масштабируемый анализ и конвейеры ML. Если вы уже знакомы с Python и Pandas, то большая часть ваших знаний может быть применена к Spark. Я показал, как выполнять некоторые общие операции с PySpark для начальной загрузки процесса обучения. Я также продемонстрировал некоторые недавние функциональные возможности Spark с пользовательскими функциями Pandas, которые позволяют выполнять код Python в распределенном режиме.Существуют отличные среды, которые позволяют легко приступить к работе с кластером Spark, что делает сейчас отличное время для изучения PySpark!

Hadoop — паркетный предикат pushdown

Переполнение стека

1. Товары
2. Клиенты
3. Случаи использования

1. Переполнение стека Публичные вопросы и ответы
2. Команды Частные вопросы и ответы для вашей команды
3. предприятие Частные вопросы и ответы для вашего предприятия
4. работы Программирование и связанные с ним технические возможности карьерного роста
5. Талант Нанимать технический талант
6. реклама Связаться с разработчиками по всему миру

Загрузка…

,