BIGTOP-1991. Add BigTop Weatherman

12 files changed, 1070 insertions, 0 deletions

diff --git a/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/WeatherGenerator.java b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/WeatherGenerator.java
new file mode 100644
index 00000000..0601c147
--- /dev/null
+++ b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/WeatherGenerator.java
@@ -0,0 +1,67 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bigtop.datagenerators.weatherman;
+
+import java.util.List;
+
+import org.apache.bigtop.datagenerators.locations.Location;
+import org.apache.bigtop.datagenerators.samplers.SeedFactory;
+import org.apache.bigtop.datagenerators.samplers.samplers.Sampler;
+import org.apache.bigtop.datagenerators.weatherman.internal.WeatherParametersReader;
+import org.apache.bigtop.datagenerators.weatherman.internal.WeatherSamplerBuilder;
+import org.apache.bigtop.datagenerators.weatherman.internal.WeatherStationParameters;
+import org.joda.time.LocalDate;
+
+/**
+ * Generates daily weather records for a given location.
+ *
+ * @author rnowling
+ *
+ */
+public class WeatherGenerator implements Sampler<WeatherRecord> {
+  private final Sampler<WeatherRecord> weatherSampler;
+
+  /**
+   * Initializes the generator.
+   *
+   * @param startDate
+   *          - first day of simulation
+   * @param location
+   *          - location to simulate weather for
+   * @param seedFactory
+   *          - for initializing seeds
+   * @throws Exception
+   *           - if weather parameter data cannot be read
+   */
+  public WeatherGenerator(LocalDate startDate, Location location,
+          SeedFactory seedFactory) throws Exception {
+    List<WeatherStationParameters> parameters = new WeatherParametersReader()
+            .readParameters();
+    WeatherSamplerBuilder builder = new WeatherSamplerBuilder(parameters,
+            location, startDate, seedFactory);
+    weatherSampler = builder.build();
+  }
+
+  /**
+   * Generates a daily WeatherRecord for consecutive days, one day per call.
+   *
+   * @return Weather data for a single day
+   * @throws Exception
+   */
+  public WeatherRecord sample() throws Exception {
+    return weatherSampler.sample();
+  }
+}
diff --git a/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/WeatherRecord.java b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/WeatherRecord.java
new file mode 100644
index 00000000..04d2eaa1
--- /dev/null
+++ b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/WeatherRecord.java
@@ -0,0 +1,77 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bigtop.datagenerators.weatherman;
+
+import org.joda.time.LocalDate;
+
+/**
+ * Weather data for a single day and location
+ *
+ * @author rnowling
+ *
+ */
+public interface WeatherRecord {
+
+  /**
+   * Day
+   *
+   * @return date
+   */
+  public abstract LocalDate getDate();
+
+  /**
+   * Average daily temperature in units of Fahrenheit
+   *
+   * @return temperature
+   */
+  public abstract double getTemperature();
+
+  /**
+   * Daily total precipitation measured in millimeters
+   *
+   * @return total precipitation
+   */
+  public abstract double getPrecipitation();
+
+  /**
+   * Daily average wind speed in meters per second.
+   *
+   * @return average wind speed
+   */
+  public abstract double getWindSpeed();
+
+  /**
+   * Daily average wind chill in Fahrenheit
+   *
+   * @return average wind chill
+   */
+  public abstract double getWindChill();
+
+  /**
+   * Daily total rainfall in millimeters
+   *
+   * @return total rainfall
+   */
+  public abstract double getRainFall();
+
+  /**
+   * Daily total snowfall in centimeters
+   *
+   * @return total snowfall
+   */
+  public abstract double getSnowFall();
+
+}
\ No newline at end of file
diff --git a/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/Driver.java b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/Driver.java
new file mode 100644
index 00000000..0bd8c41c
--- /dev/null
+++ b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/Driver.java
@@ -0,0 +1,194 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bigtop.datagenerators.weatherman.internal;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.Writer;
+import java.util.List;
+import java.util.Random;
+
+import org.apache.bigtop.datagenerators.locations.Location;
+import org.apache.bigtop.datagenerators.locations.LocationReader;
+import org.apache.bigtop.datagenerators.samplers.SeedFactory;
+import org.apache.bigtop.datagenerators.weatherman.WeatherGenerator;
+import org.apache.bigtop.datagenerators.weatherman.WeatherRecord;
+import org.joda.time.DateTime;
+import org.joda.time.LocalDate;
+import org.joda.time.format.DateTimeFormat;
+import org.joda.time.format.DateTimeFormatter;
+
+import com.google.common.collect.Lists;
+
+public class Driver {
+  List<Location> locations;
+  Location location;
+  int simulationLength;
+  LocalDate startDate;
+  SeedFactory seedFactory;
+  File outputDir;
+
+  static final int NPARAMS = 5;
+
+  private void printUsage() {
+    String usage = "BigPetStore Data Generator\n" + "\n"
+            + "Usage: java -jar bigpetstore-weather-generator.java outputDir zipcode simulationLength startDate [seed]\n"
+            + "\n" + "outputDir - (string) directory to write files\n"
+            + "zipcode - (string) location zipcode\n"
+            + "simulationLength - (int) number of days to simulate\n"
+            + "startDate - (string) simulation start date in YYYY-MM-DD format\n"
+            + "seed - (long) seed for RNG. If not given, one is randomly generated.\n";
+
+    System.out.println(usage);
+  }
+
+  private Location findLocation(String zipcode) {
+    for (Location location : locations) {
+      if (location.getZipcode().equals(zipcode))
+        return location;
+    }
+
+    return null;
+  }
+
+  /**
+   * Returns local date from string in YYYY-MM-DD format
+   *
+   * @param text
+   * @return
+   */
+  private LocalDate parseDate(String text) {
+    DateTimeFormatter dtFormatter = DateTimeFormat.forPattern("yyyy-MM-dd");
+    DateTime dt = dtFormatter.parseDateTime(text);
+
+    return dt.toLocalDate();
+  }
+
+  public void parseArgs(String[] args) {
+    if (args.length != NPARAMS && args.length != (NPARAMS - 1)) {
+      printUsage();
+      System.exit(1);
+    }
+
+    int i = -1;
+
+    outputDir = new File(args[++i]);
+    if (!outputDir.exists()) {
+      System.err.println("Given path (" + args[i] + ") does not exist.\n");
+      printUsage();
+      System.exit(1);
+    }
+
+    if (!outputDir.isDirectory()) {
+      System.err.println("Given path (" + args[i] + ") is not a directory.\n");
+      printUsage();
+      System.exit(1);
+    }
+
+    location = findLocation(args[++i]);
+    if (location == null) {
+      System.err.println(
+              "No location found for given zipcode \"" + args[i] + "\"");
+      printUsage();
+      System.exit(1);
+    }
+
+    try {
+      simulationLength = Integer.parseInt(args[++i]);
+    } catch (Exception e) {
+      System.err.println("Unable to parse '" + args[i]
+              + "' as an int for simulationLength.\n");
+      printUsage();
+      System.exit(1);
+    }
+
+    try {
+      startDate = parseDate(args[++i]);
+    } catch (Exception e) {
+      System.err.println("Unable to parse '" + args[i]
+              + "'. Expected string in 'YYYY-MM-DD' format.\n");
+      printUsage();
+      System.exit(1);
+    }
+
+    long seed = (new Random()).nextLong();
+    if (args.length == NPARAMS) {
+      try {
+        seed = Long.parseLong(args[++i]);
+      } catch (Exception e) {
+        System.err.println(
+                "Unable to parse '" + args[i] + "' as a long for the seed.\n");
+        printUsage();
+        System.exit(1);
+      }
+    }
+
+    seedFactory = new SeedFactory(seed);
+  }
+
+  private void writeWeather(Location location, List<WeatherRecord> trajectory)
+          throws Exception {
+    File outputFile = new File(outputDir.toString() + File.separator
+            + location.getZipcode() + ".txt");
+    Writer output = new BufferedWriter(new FileWriter(outputFile));
+
+    output.write(
+            "date,city,state,zipcode,temperature,windchill,windspeed,precipitation,rainfall,snowfall\n");
+    for (WeatherRecord weather : trajectory) {
+      String record = weather.getDate() + ",";
+      record += location.getCity() + ",";
+      record += location.getState() + ",";
+      record += location.getZipcode() + ",";
+      record += weather.getTemperature() + ",";
+      record += weather.getWindChill() + ",";
+      record += weather.getWindSpeed() + ",";
+      record += weather.getPrecipitation() + ",";
+      record += weather.getRainFall() + ",";
+      record += weather.getSnowFall() + "\n";
+
+      output.write(record);
+    }
+
+    output.close();
+  }
+
+  public void run(String[] args) throws Exception {
+    locations = new LocationReader().readData();
+
+    parseArgs(args);
+
+    WeatherGenerator generator = new WeatherGenerator(startDate, location,
+            seedFactory);
+
+    LocalDate date = startDate;
+    LocalDate endDate = startDate.plusDays(simulationLength);
+    List<WeatherRecord> trajectory = Lists.newArrayList();
+
+    while (date.isBefore(endDate)) {
+      WeatherRecord weather = generator.sample();
+      trajectory.add(weather);
+      date = weather.getDate();
+    }
+
+    writeWeather(location, trajectory);
+  }
+
+  public static void main(String[] args) throws Exception {
+    Driver driver = new Driver();
+    driver.run(args);
+  }
+}
diff --git a/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/PrecipitationSampler.java b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/PrecipitationSampler.java
new file mode 100644
index 00000000..118e2451
--- /dev/null
+++ b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/PrecipitationSampler.java
@@ -0,0 +1,51 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bigtop.datagenerators.weatherman.internal;
+
+import org.apache.bigtop.datagenerators.samplers.SeedFactory;
+import org.apache.bigtop.datagenerators.samplers.samplers.ConditionalSampler;
+import org.apache.bigtop.datagenerators.samplers.samplers.ExponentialSampler;
+import org.apache.bigtop.datagenerators.samplers.samplers.Sampler;
+
+public class PrecipitationSampler implements
+        ConditionalSampler<WeatherRecordBuilder, WeatherRecordBuilder> {
+  private final Sampler<Double> precipitationSampler;
+
+  public PrecipitationSampler(double averagePrecipitation,
+          SeedFactory seedFactory) {
+    precipitationSampler = new ExponentialSampler(1.0 / averagePrecipitation,
+            seedFactory);
+  }
+
+  public WeatherRecordBuilder sample(WeatherRecordBuilder record)
+          throws Exception {
+    double temp = record.getTemperature();
+    double precipitation = precipitationSampler.sample();
+    record.setPrecipitation(precipitation);
+
+    double fractionRain = 1.0
+            / (1.0 + Math.exp(-WeatherConstants.PRECIPITATION_A
+                    * (temp - WeatherConstants.PRECIPITATION_B)));
+
+    double rainfall = fractionRain * precipitation;
+    double snowfall = WeatherConstants.PRECIPITATION_TO_SNOWFALL
+            * (1.0 - fractionRain) * precipitation;
+    record.setRainFall(rainfall);
+    record.setSnowFall(snowfall);
+
+    return record;
+  }
+}
diff --git a/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/TemperatureSampler.java b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/TemperatureSampler.java
new file mode 100644
index 00000000..b64b6e0e
--- /dev/null
+++ b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/TemperatureSampler.java
@@ -0,0 +1,71 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bigtop.datagenerators.weatherman.internal;
+
+import org.apache.bigtop.datagenerators.samplers.SeedFactory;
+import org.apache.bigtop.datagenerators.samplers.samplers.ConditionalSampler;
+import org.apache.bigtop.datagenerators.samplers.samplers.GaussianSampler;
+import org.apache.bigtop.datagenerators.samplers.samplers.Sampler;
+import org.joda.time.LocalDate;
+
+public class TemperatureSampler implements
+        ConditionalSampler<WeatherRecordBuilder, WeatherRecordBuilder> {
+  final private Sampler<Double> R;
+  final private double average;
+  final private double coeffReal;
+  final private double coeffImag;
+
+  private LocalDate date;
+  private double noise;
+
+  public TemperatureSampler(LocalDate startDate, double tempAverage,
+          double tempRealCoeff, double tempImagCoeff, double tempSigma,
+          SeedFactory seedFactory) {
+    R = new GaussianSampler(0.0, tempSigma, seedFactory);
+
+    this.average = tempAverage;
+    this.coeffReal = tempRealCoeff;
+    this.coeffImag = tempImagCoeff;
+
+    date = startDate;
+    noise = 0.0;
+  }
+
+  public WeatherRecordBuilder sample(WeatherRecordBuilder weatherRecord)
+          throws Exception {
+    double temp = 0.0;
+    while (date.isEqual(weatherRecord.getDate())
+            || date.isBefore(weatherRecord.getDate())) {
+      double dayOfYear = date.getDayOfYear();
+      temp = average
+              + coeffReal * Math.cos(-2.0 * Math.PI * dayOfYear
+                      / WeatherConstants.TEMPERATURE_PERIOD)
+              + coeffImag * Math.sin(-2.0 * Math.PI * dayOfYear
+                      / WeatherConstants.TEMPERATURE_PERIOD)
+              + noise;
+
+      noise += -1.0 * noise * WeatherConstants.TEMPERATURE_GAMMA
+              * WeatherConstants.WEATHER_TIMESTEP
+              + Math.sqrt(WeatherConstants.WEATHER_TIMESTEP) * R.sample();
+
+      date = date.plusDays(WeatherConstants.WEATHER_TIMESTEP);
+    }
+
+    weatherRecord.setTemperature(temp);
+
+    return weatherRecord;
+  }
+}
diff --git a/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherConstants.java b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherConstants.java
new file mode 100644
index 00000000..83461607
--- /dev/null
+++ b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherConstants.java
@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bigtop.datagenerators.weatherman.internal;
+
+import java.io.File;
+
+public class WeatherConstants {
+  public static final File WEATHER_PARAMETERS_FILE = new File(
+          "weather_parameters.csv");
+
+  public static final double TEMPERATURE_GAMMA = 0.5; // 2 / day
+  public static final int WEATHER_TIMESTEP = 1; // days
+  public static final double TEMPERATURE_PERIOD = 365.0; // days
+
+  public static final double PRECIPITATION_A = 0.2;
+  public static final double PRECIPITATION_B = 27.0;
+  public static final double PRECIPITATION_TO_SNOWFALL = 10.0;
+
+  public static final double WIND_CHILL_PROBABILITY_A = 0.8;
+  public static final double WIND_CHILL_PROBABILITY_B = 0.5;
+  public static final double WIND_CHILL_PROBABILITY_C = 10.0; // F
+  public static final double WIND_CHILL_PROBABILITY_D = 0.2;
+
+  public static final double WIND_SPEED_PROBABILITY_A = -0.5;
+  public static final double WIND_SPEED_PROBABILITY_B = 0.8;
+  public static final double WIND_SPEED_PROBABILITY_C = 17.5; // mph
+  public static final double WIND_SPEED_PROBABILITY_D = 1.0;
+
+  public static final double SNOWFALL_PROBABILITY_A = -0.8;
+  public static final double SNOWFALL_PROBABILITY_B = 10.0;
+  public static final double SNOWFALL_PROBABILITY_C = 0.75; // in
+  public static final double SNOWFALL_PROBABILITY_D = 1.0;
+
+  public static final double RAINFALL_PROBABILITY_A = -0.6;
+  public static final double RAINFALL_PROBABILITY_B = 7.5;
+  public static final double RAINFALL_PROBABILITY_C = 0.75; // in
+  public static final double RAINFALL_PROBABILITY_D = 1.0;
+
+  public static final double M_PER_S_TO_MPH = 2.23694;
+}
diff --git a/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherParametersReader.java b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherParametersReader.java
new file mode 100644
index 00000000..e8097d66
--- /dev/null
+++ b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherParametersReader.java
@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bigtop.datagenerators.weatherman.internal;
+
+import java.io.BufferedInputStream;
+import java.io.File;
+import java.io.InputStream;
+import java.util.List;
+import java.util.Scanner;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import com.google.common.collect.Lists;
+
+public class WeatherParametersReader {
+  private InputStream getResource(File filename) throws Exception {
+    InputStream stream = getClass()
+            .getResourceAsStream("/input_data/" + filename);
+    return new BufferedInputStream(stream);
+  }
+
+  private WeatherStationParameters parseLine(String line) {
+    line = line.trim();
+    String[] cols = line.split(",");
+
+    String WBAN = cols[0];
+    String city = cols[1];
+    String state = cols[2];
+    double latitude = Double.parseDouble(cols[3]);
+    double longitude = Double.parseDouble(cols[4]);
+    Pair<Double, Double> coordinates = Pair.of(latitude, longitude);
+    double temperatureAverage = Double.parseDouble(cols[5]);
+    double temperatureRealCoeff = Double.parseDouble(cols[6]);
+    double temperatureImagCoeff = Double.parseDouble(cols[7]);
+    double temperatureDerivStd = Double.parseDouble(cols[8]);
+    double precipitationAverage = Double.parseDouble(cols[9]);
+    double windSpeedRealCoeff = Double.parseDouble(cols[10]);
+    double windSpeedImagCoeff = Double.parseDouble(cols[11]);
+    double windSpeedK = Double.parseDouble(cols[12]);
+    double windSpeedTheta = Double.parseDouble(cols[13]);
+
+    return new WeatherStationParameters(WBAN, city, state, coordinates,
+            temperatureAverage, temperatureRealCoeff, temperatureImagCoeff,
+            temperatureDerivStd, precipitationAverage, windSpeedRealCoeff,
+            windSpeedImagCoeff, windSpeedK, windSpeedTheta);
+  }
+
+  public List<WeatherStationParameters> readParameters() throws Exception {
+    InputStream inputStream = getResource(
+            WeatherConstants.WEATHER_PARAMETERS_FILE);
+    Scanner scanner = new Scanner(inputStream);
+
+    // skip header
+    scanner.nextLine();
+
+    List<WeatherStationParameters> parameterList = Lists.newArrayList();
+    while (scanner.hasNextLine()) {
+      String line = scanner.nextLine();
+      WeatherStationParameters parameters = parseLine(line);
+      parameterList.add(parameters);
+    }
+
+    scanner.close();
+
+    return parameterList;
+  }
+}
diff --git a/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherRecordBuilder.java b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherRecordBuilder.java
new file mode 100644
index 00000000..393414e7
--- /dev/null
+++ b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherRecordBuilder.java
@@ -0,0 +1,163 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bigtop.datagenerators.weatherman.internal;
+
+import java.io.Serializable;
+
+import org.apache.bigtop.datagenerators.weatherman.WeatherRecord;
+import org.joda.time.LocalDate;
+
+public class WeatherRecordBuilder implements Serializable, WeatherRecord {
+  private static final long serialVersionUID = -6397575341624071821L;
+
+  private LocalDate date;
+  private Double temperature;
+  private Double precipitation;
+  private Double windSpeed;
+  private Double windChill;
+  private Double rainFall;
+  private Double snowFall;
+
+  public WeatherRecordBuilder(LocalDate date) {
+    this.date = date;
+  }
+
+  public WeatherRecordBuilder(LocalDate date, double temperature,
+          double precipitation, double windSpeed, double windChill,
+          double rainFall, double snowFall) {
+    this.date = date;
+    this.temperature = temperature;
+    this.precipitation = precipitation;
+    this.windSpeed = windSpeed;
+    this.windChill = windChill;
+    this.rainFall = rainFall;
+    this.snowFall = snowFall;
+  }
+
+  /*
+   * (non-Javadoc)
+   *
+   * @see org.apache.bigtop.bigpetstore.datagenerator.generators.weather.
+   * WeatherRecord#getDate()
+   */
+  @Override
+  public LocalDate getDate() {
+    return date;
+  }
+
+  /*
+   * (non-Javadoc)
+   *
+   * @see org.apache.bigtop.bigpetstore.datagenerator.generators.weather.
+   * WeatherRecord#getTemperature()
+   */
+  @Override
+  public double getTemperature() {
+    return temperature;
+  }
+
+  /*
+   * (non-Javadoc)
+   *
+   * @see org.apache.bigtop.bigpetstore.datagenerator.generators.weather.
+   * WeatherRecord#getPrecipitation()
+   */
+  @Override
+  public double getPrecipitation() {
+    return precipitation;
+  }
+
+  /*
+   * (non-Javadoc)
+   *
+   * @see org.apache.bigtop.bigpetstore.datagenerator.generators.weather.
+   * WeatherRecord#getWindSpeed()
+   */
+  @Override
+  public double getWindSpeed() {
+    return windSpeed;
+  }
+
+  /*
+   * (non-Javadoc)
+   *
+   * @see org.apache.bigtop.bigpetstore.datagenerator.generators.weather.
+   * WeatherRecord#getWindChill()
+   */
+  @Override
+  public double getWindChill() {
+    return windChill;
+  }
+
+  /*
+   * (non-Javadoc)
+   *
+   * @see org.apache.bigtop.bigpetstore.datagenerator.generators.weather.
+   * WeatherRecord#getRainFall()
+   */
+  @Override
+  public double getRainFall() {
+    return rainFall;
+  }
+
+  /*
+   * (non-Javadoc)
+   *
+   * @see org.apache.bigtop.bigpetstore.datagenerator.generators.weather.
+   * WeatherRecord#getSnowFall()
+   */
+  @Override
+  public double getSnowFall() {
+    return snowFall;
+  }
+
+  public WeatherRecord build() {
+    if (temperature == null || precipitation == null || windSpeed == null
+            || windChill == null || rainFall == null || snowFall == null) {
+      throw new IllegalStateException("Not all fields have been initialized.");
+    }
+
+    return this;
+  }
+
+  public void setDate(LocalDate date) {
+    this.date = date;
+  }
+
+  public void setTemperature(Double temperature) {
+    this.temperature = temperature;
+  }
+
+  public void setPrecipitation(Double precipitation) {
+    this.precipitation = precipitation;
+  }
+
+  public void setWindSpeed(Double windSpeed) {
+    this.windSpeed = windSpeed;
+  }
+
+  public void setWindChill(Double windChill) {
+    this.windChill = windChill;
+  }
+
+  public void setRainFall(Double rainFall) {
+    this.rainFall = rainFall;
+  }
+
+  public void setSnowFall(Double snowFall) {
+    this.snowFall = snowFall;
+  }
+}
diff --git a/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherSampler.java b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherSampler.java
new file mode 100644
index 00000000..3885caa5
--- /dev/null
+++ b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherSampler.java
@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bigtop.datagenerators.weatherman.internal;
+
+import org.apache.bigtop.datagenerators.samplers.samplers.ConditionalSampler;
+import org.apache.bigtop.datagenerators.samplers.samplers.Sampler;
+import org.apache.bigtop.datagenerators.weatherman.WeatherRecord;
+import org.joda.time.LocalDate;
+
+public class WeatherSampler implements Sampler<WeatherRecord> {
+  private final ConditionalSampler<WeatherRecordBuilder, WeatherRecordBuilder> tempSampler;
+  private final ConditionalSampler<WeatherRecordBuilder, WeatherRecordBuilder> windSpeedSampler;
+  private final ConditionalSampler<WeatherRecordBuilder, WeatherRecordBuilder> precipitationSampler;
+  private LocalDate date;
+
+  public WeatherSampler(LocalDate startDate,
+          ConditionalSampler<WeatherRecordBuilder, WeatherRecordBuilder> tempSampler,
+          ConditionalSampler<WeatherRecordBuilder, WeatherRecordBuilder> windSpeedSampler,
+          ConditionalSampler<WeatherRecordBuilder, WeatherRecordBuilder> precipitationSampler) {
+    this.tempSampler = tempSampler;
+    this.windSpeedSampler = windSpeedSampler;
+    this.precipitationSampler = precipitationSampler;
+    date = startDate;
+  }
+
+  public WeatherRecord sample() throws Exception {
+    WeatherRecordBuilder weatherRecord = new WeatherRecordBuilder(date);
+
+    weatherRecord = tempSampler.sample(weatherRecord);
+    weatherRecord = windSpeedSampler.sample(weatherRecord);
+    weatherRecord = precipitationSampler.sample(weatherRecord);
+
+    date = date.plusDays(WeatherConstants.WEATHER_TIMESTEP);
+
+    return weatherRecord.build();
+  }
+}
diff --git a/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherSamplerBuilder.java b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherSamplerBuilder.java
new file mode 100644
index 00000000..40128ae9
--- /dev/null
+++ b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherSamplerBuilder.java
@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bigtop.datagenerators.weatherman.internal;
+
+import java.util.Collection;
+
+import org.apache.bigtop.datagenerators.locations.Location;
+import org.apache.bigtop.datagenerators.samplers.SeedFactory;
+import org.apache.bigtop.datagenerators.samplers.samplers.ConditionalSampler;
+import org.apache.bigtop.datagenerators.samplers.samplers.Sampler;
+import org.apache.bigtop.datagenerators.weatherman.WeatherRecord;
+import org.joda.time.LocalDate;
+
+public class WeatherSamplerBuilder {
+
+  private final WeatherStationParameters parameters;
+  private final SeedFactory seedFactory;
+  private final LocalDate startDate;
+
+  public WeatherSamplerBuilder(
+          Collection<WeatherStationParameters> weatherParameters,
+          Location location, LocalDate startDate, SeedFactory seedFactory) {
+    parameters = findClosest(weatherParameters, location);
+    this.seedFactory = seedFactory;
+    this.startDate = startDate;
+  }
+
+  private WeatherStationParameters findClosest(
+          Collection<WeatherStationParameters> weatherParameters,
+          Location location) {
+    WeatherStationParameters closestStation = null;
+    double minDist = Double.MAX_VALUE;
+
+    for (WeatherStationParameters parameters : weatherParameters) {
+      double dist = location.distance(parameters.getCoordinates());
+      if (dist < minDist) {
+        minDist = dist;
+        closestStation = parameters;
+      }
+    }
+
+    return closestStation;
+  }
+
+  private ConditionalSampler<WeatherRecordBuilder, WeatherRecordBuilder> buildTempSampler() {
+    return new TemperatureSampler(startDate, parameters.getTemperatureAverage(),
+            parameters.getTemperatureRealCoeff(),
+            parameters.getTemperatureImagCoeff(),
+            parameters.getTemperatureDerivStd(), seedFactory);
+  }
+
+  private ConditionalSampler<WeatherRecordBuilder, WeatherRecordBuilder> buildWindSpeedSampler() {
+    return new WindSpeedSampler(parameters.getWindSpeedRealCoeff(),
+            parameters.getWindSpeedImagCoeff(), parameters.getWindSpeedK(),
+            parameters.getWindSpeedTheta(), seedFactory);
+  }
+
+  private ConditionalSampler<WeatherRecordBuilder, WeatherRecordBuilder> buildPrecipitationSampler() {
+    return new PrecipitationSampler(parameters.getPrecipitationAverage(),
+            seedFactory);
+  }
+
+  public Sampler<WeatherRecord> build() {
+    return new WeatherSampler(startDate, buildTempSampler(),
+            buildWindSpeedSampler(), buildPrecipitationSampler());
+  }
+}
diff --git a/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherStationParameters.java b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherStationParameters.java
new file mode 100644
index 00000000..7edba341
--- /dev/null
+++ b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WeatherStationParameters.java
@@ -0,0 +1,110 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bigtop.datagenerators.weatherman.internal;
+
+import java.io.Serializable;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+public class WeatherStationParameters implements Serializable {
+  private static final long serialVersionUID = -7268791467819627718L;
+
+  private String WBAN;
+  private String city;
+  private String state;
+  private Pair<Double, Double> coordinates;
+  private double temperatureAverage;
+  private double temperatureRealCoeff;
+  private double temperatureImagCoeff;
+  private double temperatureDerivStd;
+  private double precipitationAverage;
+  private double windSpeedRealCoeff;
+  private double windSpeedImagCoeff;
+  private double windSpeedK;
+  private double windSpeedTheta;
+
+  public WeatherStationParameters(String WBAN, String city, String state,
+          Pair<Double, Double> coordinates, double temperatureAverage,
+          double temperatureRealCoeff, double temperatureImagCoeff,
+          double temperatureDerivStd, double precipitationAverage,
+          double windSpeedRealCoeff, double windSpeedImagCoeff,
+          double windSpeedK, double windSpeedTheta) {
+    this.city = city;
+    this.state = state;
+    this.coordinates = coordinates;
+    this.temperatureAverage = temperatureAverage;
+    this.temperatureRealCoeff = temperatureRealCoeff;
+    this.temperatureImagCoeff = temperatureImagCoeff;
+    this.temperatureDerivStd = temperatureDerivStd;
+    this.precipitationAverage = precipitationAverage;
+    this.windSpeedRealCoeff = windSpeedRealCoeff;
+    this.windSpeedImagCoeff = windSpeedImagCoeff;
+    this.windSpeedK = windSpeedK;
+    this.windSpeedTheta = windSpeedTheta;
+  }
+
+  public String getWBAN() {
+    return WBAN;
+  }
+
+  public String getCity() {
+    return city;
+  }
+
+  public String getState() {
+    return state;
+  }
+
+  public Pair<Double, Double> getCoordinates() {
+    return coordinates;
+  }
+
+  public double getTemperatureAverage() {
+    return temperatureAverage;
+  }
+
+  public double getTemperatureRealCoeff() {
+    return temperatureRealCoeff;
+  }
+
+  public double getTemperatureImagCoeff() {
+    return temperatureImagCoeff;
+  }
+
+  public double getTemperatureDerivStd() {
+    return temperatureDerivStd;
+  }
+
+  public double getPrecipitationAverage() {
+    return precipitationAverage;
+  }
+
+  public double getWindSpeedRealCoeff() {
+    return windSpeedRealCoeff;
+  }
+
+  public double getWindSpeedImagCoeff() {
+    return windSpeedImagCoeff;
+  }
+
+  public double getWindSpeedK() {
+    return windSpeedK;
+  }
+
+  public double getWindSpeedTheta() {
+    return windSpeedTheta;
+  }
+}
diff --git a/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WindSpeedSampler.java b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WindSpeedSampler.java
new file mode 100644
index 00000000..c4454325
--- /dev/null
+++ b/bigtop-data-generators/bigtop-weatherman/src/main/java/org/apache/bigtop/datagenerators/weatherman/internal/WindSpeedSampler.java
@@ -0,0 +1,74 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bigtop.datagenerators.weatherman.internal;
+
+import org.apache.bigtop.datagenerators.samplers.SeedFactory;
+import org.apache.bigtop.datagenerators.samplers.samplers.ConditionalSampler;
+import org.apache.bigtop.datagenerators.samplers.samplers.GammaSampler;
+import org.apache.bigtop.datagenerators.samplers.samplers.Sampler;
+
+public class WindSpeedSampler implements
+        ConditionalSampler<WeatherRecordBuilder, WeatherRecordBuilder> {
+  private final double coeffReal;
+  private final double coeffImag;
+  private final Sampler<Double> gamma;
+
+  public WindSpeedSampler(double windSpeedRealCoeff, double windSpeedImagCoeff,
+          double windSpeedK, double windSpeedTheta, SeedFactory seedFactory) {
+    coeffReal = windSpeedRealCoeff;
+    coeffImag = windSpeedImagCoeff;
+
+    gamma = new GammaSampler(windSpeedK, windSpeedTheta, seedFactory);
+
+  }
+
+  /**
+   *
+   * @param temp
+   *          - given in Fahrenheit
+   * @param windSpeed
+   *          - given in MPH
+   * @return
+   */
+  private Double windChill(double temp, double windSpeed) {
+    double v_16 = Math.pow(windSpeed, 0.16);
+    double windChill = 35.74 + 0.6215 * temp - 35.74 * v_16
+            + 0.4275 * temp * v_16;
+
+    return windChill;
+  }
+
+  public WeatherRecordBuilder sample(WeatherRecordBuilder weatherRecord)
+          throws Exception {
+
+    double dayOfYear = weatherRecord.getDate().getDayOfYear();
+    // meters/second
+    double windSpeed = Math.max(0.0,
+            coeffReal
+                    * Math.cos(-2.0 * Math.PI * dayOfYear
+                            / WeatherConstants.TEMPERATURE_PERIOD)
+            + coeffImag * Math.sin(-2.0 * Math.PI * dayOfYear
+                    / WeatherConstants.TEMPERATURE_PERIOD) + gamma.sample());
+
+    double windChill = windChill(weatherRecord.getTemperature(),
+            windSpeed * WeatherConstants.M_PER_S_TO_MPH);
+
+    weatherRecord.setWindSpeed(windSpeed);
+    weatherRecord.setWindChill(windChill);
+
+    return weatherRecord;
+  }
+}