http://git-wip-us.apache.org/repos/asf/mahout/blob/e0573de3/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/cassandra/CassandraDataModel.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/cassandra/CassandraDataModel.java b/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/cassandra/CassandraDataModel.java new file mode 100644 index 0000000..b220993 --- /dev/null +++ b/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/cassandra/CassandraDataModel.java @@ -0,0 +1,465 @@ +/** + * 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.mahout.cf.taste.impl.model.cassandra; + +import com.google.common.base.Preconditions; +import me.prettyprint.cassandra.model.HColumnImpl; +import me.prettyprint.cassandra.serializers.BytesArraySerializer; +import me.prettyprint.cassandra.serializers.FloatSerializer; +import me.prettyprint.cassandra.serializers.LongSerializer; +import me.prettyprint.cassandra.service.OperationType; +import me.prettyprint.hector.api.Cluster; +import me.prettyprint.hector.api.ConsistencyLevelPolicy; +import me.prettyprint.hector.api.HConsistencyLevel; +import me.prettyprint.hector.api.Keyspace; +import me.prettyprint.hector.api.beans.ColumnSlice; +import me.prettyprint.hector.api.beans.HColumn; +import me.prettyprint.hector.api.factory.HFactory; +import me.prettyprint.hector.api.mutation.Mutator; +import me.prettyprint.hector.api.query.ColumnQuery; +import me.prettyprint.hector.api.query.CountQuery; +import me.prettyprint.hector.api.query.SliceQuery; +import org.apache.mahout.cf.taste.common.NoSuchItemException; +import org.apache.mahout.cf.taste.common.NoSuchUserException; +import org.apache.mahout.cf.taste.common.Refreshable; +import org.apache.mahout.cf.taste.common.TasteException; +import org.apache.mahout.cf.taste.impl.common.Cache; +import org.apache.mahout.cf.taste.impl.common.FastIDSet; +import org.apache.mahout.cf.taste.impl.common.LongPrimitiveIterator; +import org.apache.mahout.cf.taste.impl.common.Retriever; +import org.apache.mahout.cf.taste.impl.model.GenericItemPreferenceArray; +import org.apache.mahout.cf.taste.impl.model.GenericUserPreferenceArray; +import org.apache.mahout.cf.taste.model.DataModel; +import org.apache.mahout.cf.taste.model.PreferenceArray; + +import java.io.Closeable; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.atomic.AtomicReference; + +/** + * <p>A {@link DataModel} based on a Cassandra keyspace. By default it uses keyspace "recommender" but this + * can be configured. Create the keyspace before using this class; this can be done on the Cassandra command + * line with a command linke {@code create keyspace recommender;}.</p> + * + * <p>Within the keyspace, this model uses four column families:</p> + * + * <p>First, it uses a column family called "users". This is keyed by the user ID as an 8-byte long. + * It contains a column for every preference the user expresses. The column name is item ID, again as + * an 8-byte long, and value is a floating point value represnted as an IEEE 32-bit floating poitn value.</p> + * + * <p>It uses an analogous column family called "items" for the same data, but keyed by item ID rather + * than user ID. In this column family, column names are user IDs instead.</p> + * + * <p>It uses a column family called "userIDs" as well, with an identical schema. It has one row under key + * 0. IT contains a column for every user ID in th emodel. It has no values.</p> + * + * <p>Finally it also uses an analogous column family "itemIDs" containing item IDs.</p> + * + * <p>Each of these four column families needs to be created ahead of time. Again the + * Cassandra CLI can be used to do so, with commands like {@code create column family users;}.</p> + * + * <p>Note that this thread uses a long-lived Cassandra client which will run until terminated. You + * must {@link #close()} this implementation when done or the JVM will not terminate.</p> + * + * <p>This implementation still relies heavily on reading data into memory and caching, + * as it remains too data-intensive to be effective even against Cassandra. It will take some time to + * "warm up" as the first few requests will block loading user and item data into caches. This is still going + * to send a great deal of query traffic to Cassandra. It would be advisable to employ caching wrapper + * classes in your implementation, like {@link org.apache.mahout.cf.taste.impl.recommender.CachingRecommender} + * or {@link org.apache.mahout.cf.taste.impl.similarity.CachingItemSimilarity}.</p> + */ +public final class CassandraDataModel implements DataModel, Closeable { + + /** Default Cassandra host. Default: localhost */ + private static final String DEFAULT_HOST = "localhost"; + + /** Default Cassandra port. Default: 9160 */ + private static final int DEFAULT_PORT = 9160; + + /** Default Cassandra keyspace. Default: recommender */ + private static final String DEFAULT_KEYSPACE = "recommender"; + + static final String USERS_CF = "users"; + static final String ITEMS_CF = "items"; + static final String USER_IDS_CF = "userIDs"; + static final String ITEM_IDS_CF = "itemIDs"; + private static final long ID_ROW_KEY = 0L; + private static final byte[] EMPTY = new byte[0]; + + private final Cluster cluster; + private final Keyspace keyspace; + private final Cache<Long,PreferenceArray> userCache; + private final Cache<Long,PreferenceArray> itemCache; + private final Cache<Long,FastIDSet> itemIDsFromUserCache; + private final Cache<Long,FastIDSet> userIDsFromItemCache; + private final AtomicReference<Integer> userCountCache; + private final AtomicReference<Integer> itemCountCache; + + /** + * Uses the standard Cassandra host and port (localhost:9160), and keyspace name ("recommender"). + */ + public CassandraDataModel() { + this(DEFAULT_HOST, DEFAULT_PORT, DEFAULT_KEYSPACE); + } + + /** + * @param host Cassandra server host name + * @param port Cassandra server port + * @param keyspaceName name of Cassandra keyspace to use + */ + public CassandraDataModel(String host, int port, String keyspaceName) { + + Preconditions.checkNotNull(host); + Preconditions.checkArgument(port > 0, "port must be greater then 0!"); + Preconditions.checkNotNull(keyspaceName); + + cluster = HFactory.getOrCreateCluster(CassandraDataModel.class.getSimpleName(), host + ':' + port); + keyspace = HFactory.createKeyspace(keyspaceName, cluster); + keyspace.setConsistencyLevelPolicy(new OneConsistencyLevelPolicy()); + + userCache = new Cache<>(new UserPrefArrayRetriever(), 1 << 20); + itemCache = new Cache<>(new ItemPrefArrayRetriever(), 1 << 20); + itemIDsFromUserCache = new Cache<>(new ItemIDsFromUserRetriever(), 1 << 20); + userIDsFromItemCache = new Cache<>(new UserIDsFromItemRetriever(), 1 << 20); + userCountCache = new AtomicReference<>(null); + itemCountCache = new AtomicReference<>(null); + } + + @Override + public LongPrimitiveIterator getUserIDs() { + SliceQuery<Long,Long,?> query = buildNoValueSliceQuery(USER_IDS_CF); + query.setKey(ID_ROW_KEY); + FastIDSet userIDs = new FastIDSet(); + for (HColumn<Long,?> userIDColumn : query.execute().get().getColumns()) { + userIDs.add(userIDColumn.getName()); + } + return userIDs.iterator(); + } + + @Override + public PreferenceArray getPreferencesFromUser(long userID) throws TasteException { + return userCache.get(userID); + } + + @Override + public FastIDSet getItemIDsFromUser(long userID) throws TasteException { + return itemIDsFromUserCache.get(userID); + } + + @Override + public LongPrimitiveIterator getItemIDs() { + SliceQuery<Long,Long,?> query = buildNoValueSliceQuery(ITEM_IDS_CF); + query.setKey(ID_ROW_KEY); + FastIDSet itemIDs = new FastIDSet(); + for (HColumn<Long,?> itemIDColumn : query.execute().get().getColumns()) { + itemIDs.add(itemIDColumn.getName()); + } + return itemIDs.iterator(); + } + + @Override + public PreferenceArray getPreferencesForItem(long itemID) throws TasteException { + return itemCache.get(itemID); + } + + @Override + public Float getPreferenceValue(long userID, long itemID) { + ColumnQuery<Long,Long,Float> query = + HFactory.createColumnQuery(keyspace, LongSerializer.get(), LongSerializer.get(), FloatSerializer.get()); + query.setColumnFamily(USERS_CF); + query.setKey(userID); + query.setName(itemID); + HColumn<Long,Float> column = query.execute().get(); + return column == null ? null : column.getValue(); + } + + @Override + public Long getPreferenceTime(long userID, long itemID) { + ColumnQuery<Long,Long,?> query = + HFactory.createColumnQuery(keyspace, LongSerializer.get(), LongSerializer.get(), BytesArraySerializer.get()); + query.setColumnFamily(USERS_CF); + query.setKey(userID); + query.setName(itemID); + HColumn<Long,?> result = query.execute().get(); + return result == null ? null : result.getClock(); + } + + @Override + public int getNumItems() { + Integer itemCount = itemCountCache.get(); + if (itemCount == null) { + CountQuery<Long,Long> countQuery = + HFactory.createCountQuery(keyspace, LongSerializer.get(), LongSerializer.get()); + countQuery.setKey(ID_ROW_KEY); + countQuery.setColumnFamily(ITEM_IDS_CF); + countQuery.setRange(null, null, Integer.MAX_VALUE); + itemCount = countQuery.execute().get(); + itemCountCache.set(itemCount); + } + return itemCount; + } + + @Override + public int getNumUsers() { + Integer userCount = userCountCache.get(); + if (userCount == null) { + CountQuery<Long,Long> countQuery = + HFactory.createCountQuery(keyspace, LongSerializer.get(), LongSerializer.get()); + countQuery.setKey(ID_ROW_KEY); + countQuery.setColumnFamily(USER_IDS_CF); + countQuery.setRange(null, null, Integer.MAX_VALUE); + userCount = countQuery.execute().get(); + userCountCache.set(userCount); + } + return userCount; + } + + @Override + public int getNumUsersWithPreferenceFor(long itemID) throws TasteException { + /* + CountQuery<Long,Long> query = HFactory.createCountQuery(keyspace, LongSerializer.get(), LongSerializer.get()); + query.setColumnFamily(ITEMS_CF); + query.setKey(itemID); + query.setRange(null, null, Integer.MAX_VALUE); + return query.execute().get(); + */ + return userIDsFromItemCache.get(itemID).size(); + } + + @Override + public int getNumUsersWithPreferenceFor(long itemID1, long itemID2) throws TasteException { + FastIDSet userIDs1 = userIDsFromItemCache.get(itemID1); + FastIDSet userIDs2 = userIDsFromItemCache.get(itemID2); + return userIDs1.size() < userIDs2.size() + ? userIDs2.intersectionSize(userIDs1) + : userIDs1.intersectionSize(userIDs2); + } + + @Override + public void setPreference(long userID, long itemID, float value) { + + if (Float.isNaN(value)) { + value = 1.0f; + } + + long now = System.currentTimeMillis(); + + Mutator<Long> mutator = HFactory.createMutator(keyspace, LongSerializer.get()); + + HColumn<Long,Float> itemForUsers = new HColumnImpl<>(LongSerializer.get(), FloatSerializer.get()); + itemForUsers.setName(itemID); + itemForUsers.setClock(now); + itemForUsers.setValue(value); + mutator.addInsertion(userID, USERS_CF, itemForUsers); + + HColumn<Long,Float> userForItems = new HColumnImpl<>(LongSerializer.get(), FloatSerializer.get()); + userForItems.setName(userID); + userForItems.setClock(now); + userForItems.setValue(value); + mutator.addInsertion(itemID, ITEMS_CF, userForItems); + + HColumn<Long,byte[]> userIDs = new HColumnImpl<>(LongSerializer.get(), BytesArraySerializer.get()); + userIDs.setName(userID); + userIDs.setClock(now); + userIDs.setValue(EMPTY); + mutator.addInsertion(ID_ROW_KEY, USER_IDS_CF, userIDs); + + HColumn<Long,byte[]> itemIDs = new HColumnImpl<>(LongSerializer.get(), BytesArraySerializer.get()); + itemIDs.setName(itemID); + itemIDs.setClock(now); + itemIDs.setValue(EMPTY); + mutator.addInsertion(ID_ROW_KEY, ITEM_IDS_CF, itemIDs); + + mutator.execute(); + } + + @Override + public void removePreference(long userID, long itemID) { + Mutator<Long> mutator = HFactory.createMutator(keyspace, LongSerializer.get()); + mutator.addDeletion(userID, USERS_CF, itemID, LongSerializer.get()); + mutator.addDeletion(itemID, ITEMS_CF, userID, LongSerializer.get()); + mutator.execute(); + // Not deleting from userIDs, itemIDs though + } + + /** + * @return true + */ + @Override + public boolean hasPreferenceValues() { + return true; + } + + /** + * @return Float#NaN + */ + @Override + public float getMaxPreference() { + return Float.NaN; + } + + /** + * @return Float#NaN + */ + @Override + public float getMinPreference() { + return Float.NaN; + } + + @Override + public void refresh(Collection<Refreshable> alreadyRefreshed) { + userCache.clear(); + itemCache.clear(); + userIDsFromItemCache.clear(); + itemIDsFromUserCache.clear(); + userCountCache.set(null); + itemCountCache.set(null); + } + + @Override + public String toString() { + return "CassandraDataModel[" + keyspace + ']'; + } + + @Override + public void close() { + HFactory.shutdownCluster(cluster); + } + + + private SliceQuery<Long,Long,byte[]> buildNoValueSliceQuery(String cf) { + SliceQuery<Long,Long,byte[]> query = + HFactory.createSliceQuery(keyspace, LongSerializer.get(), LongSerializer.get(), BytesArraySerializer.get()); + query.setColumnFamily(cf); + query.setRange(null, null, false, Integer.MAX_VALUE); + return query; + } + + private SliceQuery<Long,Long,Float> buildValueSliceQuery(String cf) { + SliceQuery<Long,Long,Float> query = + HFactory.createSliceQuery(keyspace, LongSerializer.get(), LongSerializer.get(), FloatSerializer.get()); + query.setColumnFamily(cf); + query.setRange(null, null, false, Integer.MAX_VALUE); + return query; + } + + + private static final class OneConsistencyLevelPolicy implements ConsistencyLevelPolicy { + @Override + public HConsistencyLevel get(OperationType op) { + return HConsistencyLevel.ONE; + } + + @Override + public HConsistencyLevel get(OperationType op, String cfName) { + return HConsistencyLevel.ONE; + } + } + + private final class UserPrefArrayRetriever implements Retriever<Long, PreferenceArray> { + @Override + public PreferenceArray get(Long userID) throws TasteException { + SliceQuery<Long,Long,Float> query = buildValueSliceQuery(USERS_CF); + query.setKey(userID); + + ColumnSlice<Long,Float> result = query.execute().get(); + if (result == null) { + throw new NoSuchUserException(userID); + } + List<HColumn<Long,Float>> itemIDColumns = result.getColumns(); + if (itemIDColumns.isEmpty()) { + throw new NoSuchUserException(userID); + } + int size = itemIDColumns.size(); + PreferenceArray prefs = new GenericUserPreferenceArray(size); + prefs.setUserID(0, userID); + for (int i = 0; i < size; i++) { + HColumn<Long,Float> itemIDColumn = itemIDColumns.get(i); + prefs.setItemID(i, itemIDColumn.getName()); + prefs.setValue(i, itemIDColumn.getValue()); + } + return prefs; + } + } + + private final class ItemPrefArrayRetriever implements Retriever<Long, PreferenceArray> { + @Override + public PreferenceArray get(Long itemID) throws TasteException { + SliceQuery<Long,Long,Float> query = buildValueSliceQuery(ITEMS_CF); + query.setKey(itemID); + ColumnSlice<Long,Float> result = query.execute().get(); + if (result == null) { + throw new NoSuchItemException(itemID); + } + List<HColumn<Long,Float>> userIDColumns = result.getColumns(); + if (userIDColumns.isEmpty()) { + throw new NoSuchItemException(itemID); + } + int size = userIDColumns.size(); + PreferenceArray prefs = new GenericItemPreferenceArray(size); + prefs.setItemID(0, itemID); + for (int i = 0; i < size; i++) { + HColumn<Long,Float> userIDColumn = userIDColumns.get(i); + prefs.setUserID(i, userIDColumn.getName()); + prefs.setValue(i, userIDColumn.getValue()); + } + return prefs; + } + } + + private final class UserIDsFromItemRetriever implements Retriever<Long, FastIDSet> { + @Override + public FastIDSet get(Long itemID) throws TasteException { + SliceQuery<Long,Long,byte[]> query = buildNoValueSliceQuery(ITEMS_CF); + query.setKey(itemID); + ColumnSlice<Long,byte[]> result = query.execute().get(); + if (result == null) { + throw new NoSuchItemException(itemID); + } + List<HColumn<Long,byte[]>> columns = result.getColumns(); + FastIDSet userIDs = new FastIDSet(columns.size()); + for (HColumn<Long,?> userIDColumn : columns) { + userIDs.add(userIDColumn.getName()); + } + return userIDs; + } + } + + private final class ItemIDsFromUserRetriever implements Retriever<Long, FastIDSet> { + @Override + public FastIDSet get(Long userID) throws TasteException { + SliceQuery<Long,Long,byte[]> query = buildNoValueSliceQuery(USERS_CF); + query.setKey(userID); + FastIDSet itemIDs = new FastIDSet(); + ColumnSlice<Long,byte[]> result = query.execute().get(); + if (result == null) { + throw new NoSuchUserException(userID); + } + List<HColumn<Long,byte[]>> columns = result.getColumns(); + if (columns.isEmpty()) { + throw new NoSuchUserException(userID); + } + for (HColumn<Long,?> itemIDColumn : columns) { + itemIDs.add(itemIDColumn.getName()); + } + return itemIDs; + } + } + +}
http://git-wip-us.apache.org/repos/asf/mahout/blob/e0573de3/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/hbase/HBaseDataModel.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/hbase/HBaseDataModel.java b/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/hbase/HBaseDataModel.java new file mode 100644 index 0000000..9735ffe --- /dev/null +++ b/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/hbase/HBaseDataModel.java @@ -0,0 +1,497 @@ +/* + * 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.mahout.cf.taste.impl.model.hbase; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.HColumnDescriptor; +import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.client.Delete; +import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.HBaseAdmin; +import org.apache.hadoop.hbase.client.HTableFactory; +import org.apache.hadoop.hbase.client.HTableInterface; +import org.apache.hadoop.hbase.client.HTablePool; +import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.ResultScanner; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.filter.FilterList; +import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter; +import org.apache.hadoop.hbase.filter.KeyOnlyFilter; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.mahout.cf.taste.common.NoSuchItemException; +import org.apache.mahout.cf.taste.common.NoSuchUserException; +import org.apache.mahout.cf.taste.common.Refreshable; +import org.apache.mahout.cf.taste.common.TasteException; +import org.apache.mahout.cf.taste.impl.common.FastIDSet; +import org.apache.mahout.cf.taste.impl.common.LongPrimitiveIterator; +import org.apache.mahout.cf.taste.impl.model.GenericItemPreferenceArray; +import org.apache.mahout.cf.taste.impl.model.GenericUserPreferenceArray; +import org.apache.mahout.cf.taste.model.DataModel; +import org.apache.mahout.cf.taste.model.PreferenceArray; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Closeable; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.SortedMap; + +/** + * <p>Naive approach of storing one preference as one value in the table. + * Preferences are indexed as (user, item) and (item, user) for O(1) lookups.</p> + * + * <p>The default table name is "taste", this can be set through a constructor + * argument. Each row has a value starting with "i" or "u" followed by the + * actual id encoded as a big endian long.</p> + * + * <p>E.g., "u\x00\x00\x00\x00\x00\x00\x04\xd2" is user 1234L</p> + * + * <p>There are two column families: "users" and "items".</p> + * + * <p>The "users" column family holds user->item preferences. Each userID is the + * column qualifier and the value is the preference.</p> + * + * <p>The "items" column fmaily holds item->user preferences. Each itemID is the + * column qualifier and the value is the preference.</p> + * + * <p>User IDs and item IDs are cached in a FastIDSet since it requires a full + * table scan to build these sets. Preferences are not cached since they + * are pretty cheap lookups in HBase (also caching the Preferences defeats + * the purpose of a scalable storage engine like HBase).</p> + */ +public final class HBaseDataModel implements DataModel, Closeable { + + private static final Logger log = LoggerFactory.getLogger(HBaseDataModel.class); + + private static final String DEFAULT_TABLE = "taste"; + private static final byte[] USERS_CF = Bytes.toBytes("users"); + private static final byte[] ITEMS_CF = Bytes.toBytes("items"); + + private final HTablePool pool; + private final String tableName; + + // Cache of user and item ids + private volatile FastIDSet itemIDs; + private volatile FastIDSet userIDs; + + public HBaseDataModel(String zkConnect) throws IOException { + this(zkConnect, DEFAULT_TABLE); + } + + public HBaseDataModel(String zkConnect, String tableName) throws IOException { + log.info("Using HBase table {}", tableName); + Configuration conf = HBaseConfiguration.create(); + conf.set("hbase.zookeeper.quorum", zkConnect); + HTableFactory tableFactory = new HTableFactory(); + this.pool = new HTablePool(conf, 8, tableFactory); + this.tableName = tableName; + + bootstrap(conf); + // Warm the cache + refresh(null); + } + + public HBaseDataModel(HTablePool pool, String tableName, Configuration conf) throws IOException { + log.info("Using HBase table {}", tableName); + this.pool = pool; + this.tableName = tableName; + + bootstrap(conf); + + // Warm the cache + refresh(null); + } + + public String getTableName() { + return tableName; + } + + /** + * Create the table if it doesn't exist + */ + private void bootstrap(Configuration conf) throws IOException { + HTableDescriptor tDesc = new HTableDescriptor(Bytes.toBytes(tableName)); + tDesc.addFamily(new HColumnDescriptor(USERS_CF)); + tDesc.addFamily(new HColumnDescriptor(ITEMS_CF)); + try (HBaseAdmin admin = new HBaseAdmin(conf)) { + admin.createTable(tDesc); + log.info("Created table {}", tableName); + } + } + + /** + * Prefix a user id with "u" and convert to byte[] + */ + private static byte[] userToBytes(long userID) { + ByteBuffer bb = ByteBuffer.allocate(9); + bb.put((byte)0x75); // The letter "u" + bb.putLong(userID); + return bb.array(); + } + + /** + * Prefix an item id with "i" and convert to byte[] + */ + private static byte[] itemToBytes(long itemID) { + ByteBuffer bb = ByteBuffer.allocate(9); + bb.put((byte)0x69); // The letter "i" + bb.putLong(itemID); + return bb.array(); + } + + /** + * Extract the id out of a prefix byte[] id + */ + private static long bytesToUserOrItemID(byte[] ba) { + ByteBuffer bb = ByteBuffer.wrap(ba); + return bb.getLong(1); + } + + /* DataModel interface */ + + @Override + public LongPrimitiveIterator getUserIDs() { + return userIDs.iterator(); + } + + @Override + public PreferenceArray getPreferencesFromUser(long userID) throws TasteException { + Result result; + try { + HTableInterface table = pool.getTable(tableName); + Get get = new Get(userToBytes(userID)); + get.addFamily(ITEMS_CF); + result = table.get(get); + table.close(); + } catch (IOException e) { + throw new TasteException("Failed to retrieve user preferences from HBase", e); + } + + if (result.isEmpty()) { + throw new NoSuchUserException(userID); + } + + SortedMap<byte[], byte[]> families = result.getFamilyMap(ITEMS_CF); + PreferenceArray prefs = new GenericUserPreferenceArray(families.size()); + prefs.setUserID(0, userID); + int i = 0; + for (Map.Entry<byte[], byte[]> entry : families.entrySet()) { + prefs.setItemID(i, Bytes.toLong(entry.getKey())); + prefs.setValue(i, Bytes.toFloat(entry.getValue())); + i++; + } + return prefs; + } + + @Override + public FastIDSet getItemIDsFromUser(long userID) throws TasteException { + Result result; + try { + HTableInterface table = pool.getTable(tableName); + Get get = new Get(userToBytes(userID)); + get.addFamily(ITEMS_CF); + result = table.get(get); + table.close(); + } catch (IOException e) { + throw new TasteException("Failed to retrieve item IDs from HBase", e); + } + + if (result.isEmpty()) { + throw new NoSuchUserException(userID); + } + + SortedMap<byte[],byte[]> families = result.getFamilyMap(ITEMS_CF); + FastIDSet ids = new FastIDSet(families.size()); + for (byte[] family : families.keySet()) { + ids.add(Bytes.toLong(family)); + } + return ids; + } + + @Override + public LongPrimitiveIterator getItemIDs() { + return itemIDs.iterator(); + } + + @Override + public PreferenceArray getPreferencesForItem(long itemID) throws TasteException { + Result result; + try { + HTableInterface table = pool.getTable(tableName); + Get get = new Get(itemToBytes(itemID)); + get.addFamily(USERS_CF); + result = table.get(get); + table.close(); + } catch (IOException e) { + throw new TasteException("Failed to retrieve item preferences from HBase", e); + } + + if (result.isEmpty()) { + throw new NoSuchItemException(itemID); + } + + SortedMap<byte[], byte[]> families = result.getFamilyMap(USERS_CF); + PreferenceArray prefs = new GenericItemPreferenceArray(families.size()); + prefs.setItemID(0, itemID); + int i = 0; + for (Map.Entry<byte[], byte[]> entry : families.entrySet()) { + prefs.setUserID(i, Bytes.toLong(entry.getKey())); + prefs.setValue(i, Bytes.toFloat(entry.getValue())); + i++; + } + return prefs; + } + + @Override + public Float getPreferenceValue(long userID, long itemID) throws TasteException { + Result result; + try { + HTableInterface table = pool.getTable(tableName); + Get get = new Get(userToBytes(userID)); + get.addColumn(ITEMS_CF, Bytes.toBytes(itemID)); + result = table.get(get); + table.close(); + } catch (IOException e) { + throw new TasteException("Failed to retrieve user preferences from HBase", e); + } + + if (result.isEmpty()) { + throw new NoSuchUserException(userID); + } + + if (result.containsColumn(ITEMS_CF, Bytes.toBytes(itemID))) { + return Bytes.toFloat(result.getValue(ITEMS_CF, Bytes.toBytes(itemID))); + } else { + return null; + } + } + + @Override + public Long getPreferenceTime(long userID, long itemID) throws TasteException { + Result result; + try { + HTableInterface table = pool.getTable(tableName); + Get get = new Get(userToBytes(userID)); + get.addColumn(ITEMS_CF, Bytes.toBytes(itemID)); + result = table.get(get); + table.close(); + } catch (IOException e) { + throw new TasteException("Failed to retrieve user preferences from HBase", e); + } + + if (result.isEmpty()) { + throw new NoSuchUserException(userID); + } + + if (result.containsColumn(ITEMS_CF, Bytes.toBytes(itemID))) { + KeyValue kv = result.getColumnLatest(ITEMS_CF, Bytes.toBytes(itemID)); + return kv.getTimestamp(); + } else { + return null; + } + } + + @Override + public int getNumItems() { + return itemIDs.size(); + } + + @Override + public int getNumUsers() { + return userIDs.size(); + } + + @Override + public int getNumUsersWithPreferenceFor(long itemID) throws TasteException { + PreferenceArray prefs = getPreferencesForItem(itemID); + return prefs.length(); + } + + @Override + public int getNumUsersWithPreferenceFor(long itemID1, long itemID2) throws TasteException { + Result[] results; + try { + HTableInterface table = pool.getTable(tableName); + List<Get> gets = new ArrayList<>(2); + gets.add(new Get(itemToBytes(itemID1))); + gets.add(new Get(itemToBytes(itemID2))); + gets.get(0).addFamily(USERS_CF); + gets.get(1).addFamily(USERS_CF); + results = table.get(gets); + table.close(); + } catch (IOException e) { + throw new TasteException("Failed to retrieve item preferences from HBase", e); + } + + if (results[0].isEmpty()) { + throw new NoSuchItemException(itemID1); + } + if (results[1].isEmpty()) { + throw new NoSuchItemException(itemID2); + } + + // First item + Result result = results[0]; + SortedMap<byte[], byte[]> families = result.getFamilyMap(USERS_CF); + FastIDSet idSet1 = new FastIDSet(families.size()); + for (byte[] id : families.keySet()) { + idSet1.add(Bytes.toLong(id)); + } + + // Second item + result = results[1]; + families = result.getFamilyMap(USERS_CF); + FastIDSet idSet2 = new FastIDSet(families.size()); + for (byte[] id : families.keySet()) { + idSet2.add(Bytes.toLong(id)); + } + + return idSet1.intersectionSize(idSet2); + } + + @Override + public void setPreference(long userID, long itemID, float value) throws TasteException { + try { + HTableInterface table = pool.getTable(tableName); + List<Put> puts = new ArrayList<>(2); + puts.add(new Put(userToBytes(userID))); + puts.add(new Put(itemToBytes(itemID))); + puts.get(0).add(ITEMS_CF, Bytes.toBytes(itemID), Bytes.toBytes(value)); + puts.get(1).add(USERS_CF, Bytes.toBytes(userID), Bytes.toBytes(value)); + table.put(puts); + table.close(); + } catch (IOException e) { + throw new TasteException("Failed to store preference in HBase", e); + } + } + + @Override + public void removePreference(long userID, long itemID) throws TasteException { + try { + HTableInterface table = pool.getTable(tableName); + List<Delete> deletes = new ArrayList<>(2); + deletes.add(new Delete(userToBytes(userID))); + deletes.add(new Delete(itemToBytes(itemID))); + deletes.get(0).deleteColumns(ITEMS_CF, Bytes.toBytes(itemID)); + deletes.get(1).deleteColumns(USERS_CF, Bytes.toBytes(userID)); + table.delete(deletes); + table.close(); + } catch (IOException e) { + throw new TasteException("Failed to remove preference from HBase", e); + } + } + + @Override + public boolean hasPreferenceValues() { + return true; + } + + @Override + public float getMaxPreference() { + throw new UnsupportedOperationException(); + } + + @Override + public float getMinPreference() { + throw new UnsupportedOperationException(); + } + + /* Closeable interface */ + + @Override + public void close() throws IOException { + pool.close(); + } + + /* Refreshable interface */ + + @Override + public void refresh(Collection<Refreshable> alreadyRefreshed) { + if (alreadyRefreshed == null || !alreadyRefreshed.contains(this)) { + try { + log.info("Refreshing item and user ID caches"); + long t1 = System.currentTimeMillis(); + refreshItemIDs(); + refreshUserIDs(); + long t2 = System.currentTimeMillis(); + log.info("Finished refreshing caches in {} ms", t2 - t1); + } catch (IOException e) { + throw new IllegalStateException("Could not reload DataModel", e); + } + } + } + + /* + * Refresh the item id cache. Warning: this does a large table scan + */ + private synchronized void refreshItemIDs() throws IOException { + // Get the list of item ids + HTableInterface table = pool.getTable(tableName); + Scan scan = new Scan(new byte[]{0x69}, new byte[]{0x70}); + scan.setFilter(new FilterList(FilterList.Operator.MUST_PASS_ALL, new KeyOnlyFilter(), new FirstKeyOnlyFilter())); + ResultScanner scanner = table.getScanner(scan); + Collection<Long> ids = new LinkedList<>(); + for (Result result : scanner) { + ids.add(bytesToUserOrItemID(result.getRow())); + } + table.close(); + + // Copy into FastIDSet + FastIDSet itemIDs = new FastIDSet(ids.size()); + for (long l : ids) { + itemIDs.add(l); + } + + // Swap with the active + this.itemIDs = itemIDs; + } + + /* + * Refresh the user id cache. Warning: this does a large table scan + */ + private synchronized void refreshUserIDs() throws IOException { + // Get the list of user ids + HTableInterface table = pool.getTable(tableName); + Scan scan = new Scan(new byte[]{0x75}, new byte[]{0x76}); + scan.setFilter(new FilterList(FilterList.Operator.MUST_PASS_ALL, new KeyOnlyFilter(), new FirstKeyOnlyFilter())); + ResultScanner scanner = table.getScanner(scan); + Collection<Long> ids = new LinkedList<>(); + for (Result result : scanner) { + ids.add(bytesToUserOrItemID(result.getRow())); + } + table.close(); + + // Copy into FastIDSet + FastIDSet userIDs = new FastIDSet(ids.size()); + for (long l : ids) { + userIDs.add(l); + } + + // Swap with the active + this.userIDs = userIDs; + } + +} http://git-wip-us.apache.org/repos/asf/mahout/blob/e0573de3/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/AbstractBooleanPrefJDBCDataModel.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/AbstractBooleanPrefJDBCDataModel.java b/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/AbstractBooleanPrefJDBCDataModel.java new file mode 100644 index 0000000..79ca1ac --- /dev/null +++ b/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/AbstractBooleanPrefJDBCDataModel.java @@ -0,0 +1,137 @@ +/** + * 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.mahout.cf.taste.impl.model.jdbc; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; + +import javax.sql.DataSource; + +import org.apache.mahout.cf.taste.common.TasteException; +import org.apache.mahout.cf.taste.impl.model.BooleanPreference; +import org.apache.mahout.cf.taste.model.Preference; +import org.apache.mahout.common.IOUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.base.Preconditions; + +public abstract class AbstractBooleanPrefJDBCDataModel extends AbstractJDBCDataModel { + + private static final Logger log = LoggerFactory.getLogger(AbstractBooleanPrefJDBCDataModel.class); + + static final String NO_SUCH_COLUMN = "NO_SUCH_COLUMN"; + + private final String setPreferenceSQL; + + protected AbstractBooleanPrefJDBCDataModel(DataSource dataSource, + String preferenceTable, + String userIDColumn, + String itemIDColumn, + String preferenceColumn, + String getPreferenceSQL, + String getPreferenceTimeSQL, + String getUserSQL, + String getAllUsersSQL, + String getNumItemsSQL, + String getNumUsersSQL, + String setPreferenceSQL, + String removePreferenceSQL, + String getUsersSQL, + String getItemsSQL, + String getPrefsForItemSQL, + String getNumPreferenceForItemSQL, + String getNumPreferenceForItemsSQL, + String getMaxPreferenceSQL, + String getMinPreferenceSQL) { + super(dataSource, + preferenceTable, + userIDColumn, + itemIDColumn, + preferenceColumn, + getPreferenceSQL, + getPreferenceTimeSQL, + getUserSQL, + getAllUsersSQL, + getNumItemsSQL, + getNumUsersSQL, + setPreferenceSQL, + removePreferenceSQL, + getUsersSQL, + getItemsSQL, + getPrefsForItemSQL, + getNumPreferenceForItemSQL, + getNumPreferenceForItemsSQL, + getMaxPreferenceSQL, + getMinPreferenceSQL); + this.setPreferenceSQL = setPreferenceSQL; + } + + @Override + protected Preference buildPreference(ResultSet rs) throws SQLException { + return new BooleanPreference(getLongColumn(rs, 1), getLongColumn(rs, 2)); + } + + @Override + String getSetPreferenceSQL() { + return setPreferenceSQL; + } + + @Override + public void setPreference(long userID, long itemID, float value) throws TasteException { + Preconditions.checkArgument(!Float.isNaN(value), "NaN value"); + log.debug("Setting preference for user {}, item {}", userID, itemID); + + Connection conn = null; + PreparedStatement stmt = null; + + try { + conn = getDataSource().getConnection(); + stmt = conn.prepareStatement(setPreferenceSQL); + setLongParameter(stmt, 1, userID); + setLongParameter(stmt, 2, itemID); + + log.debug("Executing SQL update: {}", setPreferenceSQL); + stmt.executeUpdate(); + + } catch (SQLException sqle) { + log.warn("Exception while setting preference", sqle); + throw new TasteException(sqle); + } finally { + IOUtils.quietClose(null, stmt, conn); + } + } + + @Override + public boolean hasPreferenceValues() { + return false; + } + + @Override + public float getMaxPreference() { + return 1.0f; + } + + @Override + public float getMinPreference() { + return 1.0f; + } + +} http://git-wip-us.apache.org/repos/asf/mahout/blob/e0573de3/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/AbstractJDBCDataModel.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/AbstractJDBCDataModel.java b/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/AbstractJDBCDataModel.java new file mode 100644 index 0000000..66f0a77 --- /dev/null +++ b/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/AbstractJDBCDataModel.java @@ -0,0 +1,787 @@ +/** + * 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.mahout.cf.taste.impl.model.jdbc; + +import com.google.common.base.Preconditions; +import org.apache.mahout.cf.taste.common.NoSuchItemException; +import org.apache.mahout.cf.taste.common.NoSuchUserException; +import org.apache.mahout.cf.taste.common.Refreshable; +import org.apache.mahout.cf.taste.common.TasteException; +import org.apache.mahout.cf.taste.impl.common.Cache; +import org.apache.mahout.cf.taste.impl.common.FastByIDMap; +import org.apache.mahout.cf.taste.impl.common.FastIDSet; +import org.apache.mahout.cf.taste.impl.common.LongPrimitiveIterator; +import org.apache.mahout.cf.taste.impl.common.Retriever; +import org.apache.mahout.cf.taste.impl.common.jdbc.AbstractJDBCComponent; +import org.apache.mahout.cf.taste.impl.common.jdbc.ResultSetIterator; +import org.apache.mahout.cf.taste.impl.model.GenericItemPreferenceArray; +import org.apache.mahout.cf.taste.impl.model.GenericPreference; +import org.apache.mahout.cf.taste.impl.model.GenericUserPreferenceArray; +import org.apache.mahout.cf.taste.model.JDBCDataModel; +import org.apache.mahout.cf.taste.model.Preference; +import org.apache.mahout.cf.taste.model.PreferenceArray; +import org.apache.mahout.common.IOUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import javax.sql.DataSource; + +/** + * <p> + * An abstract superclass for {@link JDBCDataModel} implementations, providing most of the common + * functionality that any such implementation would need. + * </p> + * + * <p> + * Performance will be a concern with any {@link JDBCDataModel}. There are going to be lots of + * simultaneous reads and some writes to one table. Make sure the table is set up optimally -- for example, + * you'll want to establish indexes. + * </p> + * + * <p> + * You'll also want to use connection pooling of some kind. Most J2EE containers like Tomcat provide + * connection pooling, so make sure the {@link DataSource} it exposes is using pooling. Outside a J2EE + * container, you can use packages like Jakarta's <a href="http://jakarta.apache.org/commons/dbcp/";>DBCP</a> + * to create a {@link DataSource} on top of your database whose {@link Connection}s are pooled. + * </p> + */ +public abstract class AbstractJDBCDataModel extends AbstractJDBCComponent implements JDBCDataModel { + + private static final Logger log = LoggerFactory.getLogger(AbstractJDBCDataModel.class); + + public static final String DEFAULT_PREFERENCE_TABLE = "taste_preferences"; + public static final String DEFAULT_USER_ID_COLUMN = "user_id"; + public static final String DEFAULT_ITEM_ID_COLUMN = "item_id"; + public static final String DEFAULT_PREFERENCE_COLUMN = "preference"; + public static final String DEFAULT_PREFERENCE_TIME_COLUMN = "timestamp"; + + private final DataSource dataSource; + private final String preferenceTable; + private final String userIDColumn; + private final String itemIDColumn; + private final String preferenceColumn; + private final String getPreferenceSQL; + private final String getPreferenceTimeSQL; + private final String getUserSQL; + private final String getAllUsersSQL; + private final String getNumItemsSQL; + private final String getNumUsersSQL; + private final String setPreferenceSQL; + private final String removePreferenceSQL; + private final String getUsersSQL; + private final String getItemsSQL; + private final String getPrefsForItemSQL; + private final String getNumPreferenceForItemsSQL; + private final String getMaxPreferenceSQL; + private final String getMinPreferenceSQL; + private int cachedNumUsers; + private int cachedNumItems; + private final Cache<Long,Integer> itemPrefCounts; + private float maxPreference; + private float minPreference; + + protected AbstractJDBCDataModel(DataSource dataSource, + String getPreferenceSQL, + String getPreferenceTimeSQL, + String getUserSQL, + String getAllUsersSQL, + String getNumItemsSQL, + String getNumUsersSQL, + String setPreferenceSQL, + String removePreferenceSQL, + String getUsersSQL, + String getItemsSQL, + String getPrefsForItemSQL, + String getNumPreferenceForItemSQL, + String getNumPreferenceForItemsSQL, + String getMaxPreferenceSQL, + String getMinPreferenceSQL) { + this(dataSource, + DEFAULT_PREFERENCE_TABLE, + DEFAULT_USER_ID_COLUMN, + DEFAULT_ITEM_ID_COLUMN, + DEFAULT_PREFERENCE_COLUMN, + getPreferenceSQL, + getPreferenceTimeSQL, + getUserSQL, + getAllUsersSQL, + getNumItemsSQL, + getNumUsersSQL, + setPreferenceSQL, + removePreferenceSQL, + getUsersSQL, + getItemsSQL, + getPrefsForItemSQL, + getNumPreferenceForItemSQL, + getNumPreferenceForItemsSQL, + getMaxPreferenceSQL, + getMinPreferenceSQL); + } + + protected AbstractJDBCDataModel(DataSource dataSource, + String preferenceTable, + String userIDColumn, + String itemIDColumn, + String preferenceColumn, + String getPreferenceSQL, + String getPreferenceTimeSQL, + String getUserSQL, + String getAllUsersSQL, + String getNumItemsSQL, + String getNumUsersSQL, + String setPreferenceSQL, + String removePreferenceSQL, + String getUsersSQL, + String getItemsSQL, + String getPrefsForItemSQL, + String getNumPreferenceForItemSQL, + String getNumPreferenceForItemsSQL, + String getMaxPreferenceSQL, + String getMinPreferenceSQL) { + + log.debug("Creating AbstractJDBCModel..."); + + AbstractJDBCComponent.checkNotNullAndLog("preferenceTable", preferenceTable); + AbstractJDBCComponent.checkNotNullAndLog("userIDColumn", userIDColumn); + AbstractJDBCComponent.checkNotNullAndLog("itemIDColumn", itemIDColumn); + AbstractJDBCComponent.checkNotNullAndLog("preferenceColumn", preferenceColumn); + + AbstractJDBCComponent.checkNotNullAndLog("dataSource", dataSource); + AbstractJDBCComponent.checkNotNullAndLog("getUserSQL", getUserSQL); + AbstractJDBCComponent.checkNotNullAndLog("getAllUsersSQL", getAllUsersSQL); + AbstractJDBCComponent.checkNotNullAndLog("getPreferenceSQL", getPreferenceSQL); + // getPreferenceTimeSQL can be null + AbstractJDBCComponent.checkNotNullAndLog("getNumItemsSQL", getNumItemsSQL); + AbstractJDBCComponent.checkNotNullAndLog("getNumUsersSQL", getNumUsersSQL); + AbstractJDBCComponent.checkNotNullAndLog("setPreferenceSQL", setPreferenceSQL); + AbstractJDBCComponent.checkNotNullAndLog("removePreferenceSQL", removePreferenceSQL); + AbstractJDBCComponent.checkNotNullAndLog("getUsersSQL", getUsersSQL); + AbstractJDBCComponent.checkNotNullAndLog("getItemsSQL", getItemsSQL); + AbstractJDBCComponent.checkNotNullAndLog("getPrefsForItemSQL", getPrefsForItemSQL); + AbstractJDBCComponent.checkNotNullAndLog("getNumPreferenceForItemSQL", getNumPreferenceForItemSQL); + AbstractJDBCComponent.checkNotNullAndLog("getNumPreferenceForItemsSQL", getNumPreferenceForItemsSQL); + AbstractJDBCComponent.checkNotNullAndLog("getMaxPreferenceSQL", getMaxPreferenceSQL); + AbstractJDBCComponent.checkNotNullAndLog("getMinPreferenceSQL", getMinPreferenceSQL); + + if (!(dataSource instanceof ConnectionPoolDataSource)) { + log.warn("You are not using ConnectionPoolDataSource. Make sure your DataSource pools connections " + + "to the database itself, or database performance will be severely reduced."); + } + + this.preferenceTable = preferenceTable; + this.userIDColumn = userIDColumn; + this.itemIDColumn = itemIDColumn; + this.preferenceColumn = preferenceColumn; + + this.dataSource = dataSource; + this.getPreferenceSQL = getPreferenceSQL; + this.getPreferenceTimeSQL = getPreferenceTimeSQL; + this.getUserSQL = getUserSQL; + this.getAllUsersSQL = getAllUsersSQL; + this.getNumItemsSQL = getNumItemsSQL; + this.getNumUsersSQL = getNumUsersSQL; + this.setPreferenceSQL = setPreferenceSQL; + this.removePreferenceSQL = removePreferenceSQL; + this.getUsersSQL = getUsersSQL; + this.getItemsSQL = getItemsSQL; + this.getPrefsForItemSQL = getPrefsForItemSQL; + //this.getNumPreferenceForItemSQL = getNumPreferenceForItemSQL; + this.getNumPreferenceForItemsSQL = getNumPreferenceForItemsSQL; + this.getMaxPreferenceSQL = getMaxPreferenceSQL; + this.getMinPreferenceSQL = getMinPreferenceSQL; + + this.cachedNumUsers = -1; + this.cachedNumItems = -1; + this.itemPrefCounts = new Cache<>(new ItemPrefCountRetriever(getNumPreferenceForItemSQL)); + + this.maxPreference = Float.NaN; + this.minPreference = Float.NaN; + } + + /** @return the {@link DataSource} that this instance is using */ + @Override + public DataSource getDataSource() { + return dataSource; + } + + public String getPreferenceTable() { + return preferenceTable; + } + + public String getUserIDColumn() { + return userIDColumn; + } + + public String getItemIDColumn() { + return itemIDColumn; + } + + public String getPreferenceColumn() { + return preferenceColumn; + } + + String getSetPreferenceSQL() { + return setPreferenceSQL; + } + + @Override + public LongPrimitiveIterator getUserIDs() throws TasteException { + log.debug("Retrieving all users..."); + try { + return new ResultSetIDIterator(getUsersSQL); + } catch (SQLException sqle) { + throw new TasteException(sqle); + } + } + + /** + * @throws NoSuchUserException + * if there is no such user + */ + @Override + public PreferenceArray getPreferencesFromUser(long userID) throws TasteException { + + log.debug("Retrieving user ID '{}'", userID); + + Connection conn = null; + PreparedStatement stmt = null; + ResultSet rs = null; + + try { + conn = dataSource.getConnection(); + stmt = conn.prepareStatement(getUserSQL, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); + stmt.setFetchDirection(ResultSet.FETCH_FORWARD); + stmt.setFetchSize(getFetchSize()); + setLongParameter(stmt, 1, userID); + + log.debug("Executing SQL query: {}", getUserSQL); + rs = stmt.executeQuery(); + + List<Preference> prefs = new ArrayList<>(); + while (rs.next()) { + prefs.add(buildPreference(rs)); + } + + if (prefs.isEmpty()) { + throw new NoSuchUserException(userID); + } + + return new GenericUserPreferenceArray(prefs); + + } catch (SQLException sqle) { + log.warn("Exception while retrieving user", sqle); + throw new TasteException(sqle); + } finally { + IOUtils.quietClose(rs, stmt, conn); + } + + } + + @Override + public FastByIDMap<PreferenceArray> exportWithPrefs() throws TasteException { + log.debug("Exporting all data"); + + Connection conn = null; + Statement stmt = null; + ResultSet rs = null; + + FastByIDMap<PreferenceArray> result = new FastByIDMap<>(); + + try { + conn = dataSource.getConnection(); + stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); + stmt.setFetchDirection(ResultSet.FETCH_FORWARD); + stmt.setFetchSize(getFetchSize()); + + log.debug("Executing SQL query: {}", getAllUsersSQL); + rs = stmt.executeQuery(getAllUsersSQL); + + Long currentUserID = null; + List<Preference> currentPrefs = new ArrayList<>(); + while (rs.next()) { + long nextUserID = getLongColumn(rs, 1); + if (currentUserID != null && !currentUserID.equals(nextUserID) && !currentPrefs.isEmpty()) { + result.put(currentUserID, new GenericUserPreferenceArray(currentPrefs)); + currentPrefs.clear(); + } + currentPrefs.add(buildPreference(rs)); + currentUserID = nextUserID; + } + if (!currentPrefs.isEmpty()) { + result.put(currentUserID, new GenericUserPreferenceArray(currentPrefs)); + } + + return result; + + } catch (SQLException sqle) { + log.warn("Exception while exporting all data", sqle); + throw new TasteException(sqle); + } finally { + IOUtils.quietClose(rs, stmt, conn); + + } + } + + @Override + public FastByIDMap<FastIDSet> exportWithIDsOnly() throws TasteException { + log.debug("Exporting all data"); + + Connection conn = null; + Statement stmt = null; + ResultSet rs = null; + + FastByIDMap<FastIDSet> result = new FastByIDMap<>(); + + try { + conn = dataSource.getConnection(); + stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); + stmt.setFetchDirection(ResultSet.FETCH_FORWARD); + stmt.setFetchSize(getFetchSize()); + + log.debug("Executing SQL query: {}", getAllUsersSQL); + rs = stmt.executeQuery(getAllUsersSQL); + + boolean currentUserIDSet = false; + long currentUserID = 0L; // value isn't used + FastIDSet currentItemIDs = new FastIDSet(2); + while (rs.next()) { + long nextUserID = getLongColumn(rs, 1); + if (currentUserIDSet && currentUserID != nextUserID && !currentItemIDs.isEmpty()) { + result.put(currentUserID, currentItemIDs); + currentItemIDs = new FastIDSet(2); + } + currentItemIDs.add(getLongColumn(rs, 2)); + currentUserID = nextUserID; + currentUserIDSet = true; + } + if (!currentItemIDs.isEmpty()) { + result.put(currentUserID, currentItemIDs); + } + + return result; + + } catch (SQLException sqle) { + log.warn("Exception while exporting all data", sqle); + throw new TasteException(sqle); + } finally { + IOUtils.quietClose(rs, stmt, conn); + + } + } + + /** + * @throws NoSuchUserException + * if there is no such user + */ + @Override + public FastIDSet getItemIDsFromUser(long userID) throws TasteException { + + log.debug("Retrieving items for user ID '{}'", userID); + + Connection conn = null; + PreparedStatement stmt = null; + ResultSet rs = null; + + try { + conn = dataSource.getConnection(); + stmt = conn.prepareStatement(getUserSQL, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); + stmt.setFetchDirection(ResultSet.FETCH_FORWARD); + stmt.setFetchSize(getFetchSize()); + setLongParameter(stmt, 1, userID); + + log.debug("Executing SQL query: {}", getUserSQL); + rs = stmt.executeQuery(); + + FastIDSet result = new FastIDSet(); + while (rs.next()) { + result.add(getLongColumn(rs, 2)); + } + + if (result.isEmpty()) { + throw new NoSuchUserException(userID); + } + + return result; + + } catch (SQLException sqle) { + log.warn("Exception while retrieving item s", sqle); + throw new TasteException(sqle); + } finally { + IOUtils.quietClose(rs, stmt, conn); + } + + } + + @Override + public Float getPreferenceValue(long userID, long itemID) throws TasteException { + log.debug("Retrieving preferences for item ID '{}'", itemID); + Connection conn = null; + PreparedStatement stmt = null; + ResultSet rs = null; + try { + conn = dataSource.getConnection(); + stmt = conn.prepareStatement(getPreferenceSQL, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); + stmt.setFetchDirection(ResultSet.FETCH_FORWARD); + stmt.setFetchSize(1); + setLongParameter(stmt, 1, userID); + setLongParameter(stmt, 2, itemID); + + log.debug("Executing SQL query: {}", getPreferenceSQL); + rs = stmt.executeQuery(); + if (rs.next()) { + return rs.getFloat(1); + } else { + return null; + } + } catch (SQLException sqle) { + log.warn("Exception while retrieving prefs for item", sqle); + throw new TasteException(sqle); + } finally { + IOUtils.quietClose(rs, stmt, conn); + } + } + + @Override + public Long getPreferenceTime(long userID, long itemID) throws TasteException { + if (getPreferenceTimeSQL == null) { + return null; + } + log.debug("Retrieving preference time for item ID '{}'", itemID); + Connection conn = null; + PreparedStatement stmt = null; + ResultSet rs = null; + try { + conn = dataSource.getConnection(); + stmt = conn.prepareStatement(getPreferenceTimeSQL, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); + stmt.setFetchDirection(ResultSet.FETCH_FORWARD); + stmt.setFetchSize(1); + setLongParameter(stmt, 1, userID); + setLongParameter(stmt, 2, itemID); + + log.debug("Executing SQL query: {}", getPreferenceTimeSQL); + rs = stmt.executeQuery(); + if (rs.next()) { + return rs.getLong(1); + } else { + return null; + } + } catch (SQLException sqle) { + log.warn("Exception while retrieving time for item", sqle); + throw new TasteException(sqle); + } finally { + IOUtils.quietClose(rs, stmt, conn); + } + } + + @Override + public LongPrimitiveIterator getItemIDs() throws TasteException { + log.debug("Retrieving all items..."); + try { + return new ResultSetIDIterator(getItemsSQL); + } catch (SQLException sqle) { + throw new TasteException(sqle); + } + } + + @Override + public PreferenceArray getPreferencesForItem(long itemID) throws TasteException { + List<Preference> list = doGetPreferencesForItem(itemID); + if (list.isEmpty()) { + throw new NoSuchItemException(itemID); + } + return new GenericItemPreferenceArray(list); + } + + protected List<Preference> doGetPreferencesForItem(long itemID) throws TasteException { + log.debug("Retrieving preferences for item ID '{}'", itemID); + Connection conn = null; + PreparedStatement stmt = null; + ResultSet rs = null; + try { + conn = dataSource.getConnection(); + stmt = conn.prepareStatement(getPrefsForItemSQL, ResultSet.TYPE_FORWARD_ONLY, + ResultSet.CONCUR_READ_ONLY); + stmt.setFetchDirection(ResultSet.FETCH_FORWARD); + stmt.setFetchSize(getFetchSize()); + setLongParameter(stmt, 1, itemID); + + log.debug("Executing SQL query: {}", getPrefsForItemSQL); + rs = stmt.executeQuery(); + List<Preference> prefs = new ArrayList<>(); + while (rs.next()) { + prefs.add(buildPreference(rs)); + } + return prefs; + } catch (SQLException sqle) { + log.warn("Exception while retrieving prefs for item", sqle); + throw new TasteException(sqle); + } finally { + IOUtils.quietClose(rs, stmt, conn); + } + } + + @Override + public int getNumItems() throws TasteException { + if (cachedNumItems < 0) { + cachedNumItems = getNumThings("items", getNumItemsSQL); + } + return cachedNumItems; + } + + @Override + public int getNumUsers() throws TasteException { + if (cachedNumUsers < 0) { + cachedNumUsers = getNumThings("users", getNumUsersSQL); + } + return cachedNumUsers; + } + + @Override + public int getNumUsersWithPreferenceFor(long itemID) throws TasteException { + return itemPrefCounts.get(itemID); + } + + @Override + public int getNumUsersWithPreferenceFor(long itemID1, long itemID2) throws TasteException { + return getNumThings("user preferring items", getNumPreferenceForItemsSQL, itemID1, itemID2); + } + + private int getNumThings(String name, String sql, long... args) throws TasteException { + log.debug("Retrieving number of {} in model", name); + Connection conn = null; + PreparedStatement stmt = null; + ResultSet rs = null; + try { + conn = dataSource.getConnection(); + stmt = conn.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); + stmt.setFetchDirection(ResultSet.FETCH_FORWARD); + stmt.setFetchSize(getFetchSize()); + if (args != null) { + for (int i = 1; i <= args.length; i++) { + setLongParameter(stmt, i, args[i - 1]); + } + } + log.debug("Executing SQL query: {}", sql); + rs = stmt.executeQuery(); + rs.next(); + return rs.getInt(1); + } catch (SQLException sqle) { + log.warn("Exception while retrieving number of {}", name, sqle); + throw new TasteException(sqle); + } finally { + IOUtils.quietClose(rs, stmt, conn); + } + } + + @Override + public void setPreference(long userID, long itemID, float value) throws TasteException { + Preconditions.checkArgument(!Float.isNaN(value), "NaN value"); + + log.debug("Setting preference for user {}, item {}", userID, itemID); + + Connection conn = null; + PreparedStatement stmt = null; + + try { + conn = dataSource.getConnection(); + stmt = conn.prepareStatement(setPreferenceSQL); + setLongParameter(stmt, 1, userID); + setLongParameter(stmt, 2, itemID); + stmt.setDouble(3, value); + stmt.setDouble(4, value); + + log.debug("Executing SQL update: {}", setPreferenceSQL); + stmt.executeUpdate(); + + } catch (SQLException sqle) { + log.warn("Exception while setting preference", sqle); + throw new TasteException(sqle); + } finally { + IOUtils.quietClose(null, stmt, conn); + } + } + + @Override + public void removePreference(long userID, long itemID) throws TasteException { + + log.debug("Removing preference for user '{}', item '{}'", userID, itemID); + + Connection conn = null; + PreparedStatement stmt = null; + + try { + conn = dataSource.getConnection(); + stmt = conn.prepareStatement(removePreferenceSQL); + setLongParameter(stmt, 1, userID); + setLongParameter(stmt, 2, itemID); + + log.debug("Executing SQL update: {}", removePreferenceSQL); + stmt.executeUpdate(); + + } catch (SQLException sqle) { + log.warn("Exception while removing preference", sqle); + throw new TasteException(sqle); + } finally { + IOUtils.quietClose(null, stmt, conn); + } + } + + @Override + public void refresh(Collection<Refreshable> alreadyRefreshed) { + cachedNumUsers = -1; + cachedNumItems = -1; + minPreference = Float.NaN; + maxPreference = Float.NaN; + itemPrefCounts.clear(); + } + + @Override + public boolean hasPreferenceValues() { + return true; + } + + @Override + public float getMaxPreference() { + if (Float.isNaN(maxPreference)) { + Connection conn = null; + PreparedStatement stmt = null; + ResultSet rs = null; + try { + conn = dataSource.getConnection(); + stmt = conn.prepareStatement(getMaxPreferenceSQL); + + log.debug("Executing SQL query: {}", getMaxPreferenceSQL); + rs = stmt.executeQuery(); + rs.next(); + maxPreference = rs.getFloat(1); + + } catch (SQLException sqle) { + log.warn("Exception while removing preference", sqle); + // do nothing + } finally { + IOUtils.quietClose(rs, stmt, conn); + } + } + return maxPreference; + } + + @Override + public float getMinPreference() { + if (Float.isNaN(minPreference)) { + Connection conn = null; + PreparedStatement stmt = null; + ResultSet rs = null; + try { + conn = dataSource.getConnection(); + stmt = conn.prepareStatement(getMinPreferenceSQL); + + log.debug("Executing SQL query: {}", getMinPreferenceSQL); + rs = stmt.executeQuery(); + rs.next(); + minPreference = rs.getFloat(1); + + } catch (SQLException sqle) { + log.warn("Exception while removing preference", sqle); + // do nothing + } finally { + IOUtils.quietClose(rs, stmt, conn); + } + } + return minPreference; + } + + // Some overrideable methods to customize the class behavior: + + protected Preference buildPreference(ResultSet rs) throws SQLException { + return new GenericPreference(getLongColumn(rs, 1), getLongColumn(rs, 2), rs.getFloat(3)); + } + + /** + * Subclasses may wish to override this if ID values in the file are not numeric. This provides a hook by + * which subclasses can inject an {@link org.apache.mahout.cf.taste.model.IDMigrator} to perform + * translation. + */ + protected long getLongColumn(ResultSet rs, int position) throws SQLException { + return rs.getLong(position); + } + + /** + * Subclasses may wish to override this if ID values in the file are not numeric. This provides a hook by + * which subclasses can inject an {@link org.apache.mahout.cf.taste.model.IDMigrator} to perform + * translation. + */ + protected void setLongParameter(PreparedStatement stmt, int position, long value) throws SQLException { + stmt.setLong(position, value); + } + + /** + * <p> + * An {@link java.util.Iterator} which returns items from a {@link ResultSet}. This is a useful way to + * iterate over all user data since it does not require all data to be read into memory at once. It does + * however require that the DB connection be held open. Note that this class will only release database + * resources after {@link #hasNext()} has been called and has returned {@code false}; callers should + * make sure to "drain" the entire set of data to avoid tying up database resources. + * </p> + */ + private final class ResultSetIDIterator extends ResultSetIterator<Long> implements LongPrimitiveIterator { + + private ResultSetIDIterator(String sql) throws SQLException { + super(dataSource, sql); + } + + @Override + protected Long parseElement(ResultSet resultSet) throws SQLException { + return getLongColumn(resultSet, 1); + } + + @Override + public long nextLong() { + return next(); + } + + /** + * @throws UnsupportedOperationException + */ + @Override + public long peek() { + // This could be supported; is it worth it? + throw new UnsupportedOperationException(); + } + } + + private final class ItemPrefCountRetriever implements Retriever<Long,Integer> { + private final String getNumPreferenceForItemSQL; + + private ItemPrefCountRetriever(String getNumPreferenceForItemSQL) { + this.getNumPreferenceForItemSQL = getNumPreferenceForItemSQL; + } + + @Override + public Integer get(Long key) throws TasteException { + return getNumThings("user preferring item", getNumPreferenceForItemSQL, key); + } + } +} http://git-wip-us.apache.org/repos/asf/mahout/blob/e0573de3/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/ConnectionPoolDataSource.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/ConnectionPoolDataSource.java b/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/ConnectionPoolDataSource.java new file mode 100644 index 0000000..ff7f661 --- /dev/null +++ b/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/ConnectionPoolDataSource.java @@ -0,0 +1,122 @@ +/** + * 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.mahout.cf.taste.impl.model.jdbc; + +import java.io.PrintWriter; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; +import java.util.logging.Logger; + +import javax.sql.DataSource; + +import org.apache.commons.dbcp.ConnectionFactory; +import org.apache.commons.dbcp.PoolableConnectionFactory; +import org.apache.commons.dbcp.PoolingDataSource; +import org.apache.commons.pool.impl.GenericObjectPool; + +import com.google.common.base.Preconditions; + +/** + * <p> + * A wrapper {@link DataSource} which pools connections. + * </p> + */ +public final class ConnectionPoolDataSource implements DataSource { + + private final DataSource delegate; + + public ConnectionPoolDataSource(DataSource underlyingDataSource) { + Preconditions.checkNotNull(underlyingDataSource); + ConnectionFactory connectionFactory = new ConfiguringConnectionFactory(underlyingDataSource); + GenericObjectPool objectPool = new GenericObjectPool(); + objectPool.setTestOnBorrow(false); + objectPool.setTestOnReturn(false); + objectPool.setTestWhileIdle(true); + objectPool.setTimeBetweenEvictionRunsMillis(60 * 1000L); + // Constructor actually sets itself as factory on pool + new PoolableConnectionFactory(connectionFactory, objectPool, null, "SELECT 1", false, false); + delegate = new PoolingDataSource(objectPool); + } + + @Override + public Connection getConnection() throws SQLException { + return delegate.getConnection(); + } + + @Override + public Connection getConnection(String username, String password) throws SQLException { + return delegate.getConnection(username, password); + } + + @Override + public PrintWriter getLogWriter() throws SQLException { + return delegate.getLogWriter(); + } + + @Override + public void setLogWriter(PrintWriter printWriter) throws SQLException { + delegate.setLogWriter(printWriter); + } + + @Override + public void setLoginTimeout(int timeout) throws SQLException { + delegate.setLoginTimeout(timeout); + } + + @Override + public int getLoginTimeout() throws SQLException { + return delegate.getLoginTimeout(); + } + + @Override + public <T> T unwrap(Class<T> iface) throws SQLException { + return delegate.unwrap(iface); + } + + @Override + public boolean isWrapperFor(Class<?> iface) throws SQLException { + return delegate.isWrapperFor(iface); + } + + // This exists for compatibility with Java 7 / JDBC 4.1, but doesn't exist + // in Java 6. In Java 7 it would @Override, but not in 6. + // @Override + public Logger getParentLogger() throws SQLFeatureNotSupportedException { + throw new SQLFeatureNotSupportedException(); + } + + private static class ConfiguringConnectionFactory implements ConnectionFactory { + + private final DataSource underlyingDataSource; + + ConfiguringConnectionFactory(DataSource underlyingDataSource) { + this.underlyingDataSource = underlyingDataSource; + } + + @Override + public Connection createConnection() throws SQLException { + Connection connection = underlyingDataSource.getConnection(); + connection.setTransactionIsolation(Connection.TRANSACTION_READ_UNCOMMITTED); + connection.setHoldability(ResultSet.CLOSE_CURSORS_AT_COMMIT); + return connection; + } + } + +} http://git-wip-us.apache.org/repos/asf/mahout/blob/e0573de3/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/GenericJDBCDataModel.java ---------------------------------------------------------------------- diff --git a/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/GenericJDBCDataModel.java b/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/GenericJDBCDataModel.java new file mode 100644 index 0000000..5dd0be9 --- /dev/null +++ b/community/mahout-mr/integration/src/main/java/org/apache/mahout/cf/taste/impl/model/jdbc/GenericJDBCDataModel.java @@ -0,0 +1,146 @@ +/** + * 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.mahout.cf.taste.impl.model.jdbc; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.util.Properties; + +import com.google.common.io.Closeables; +import org.apache.mahout.cf.taste.common.TasteException; +import org.apache.mahout.cf.taste.impl.common.jdbc.AbstractJDBCComponent; + +/** + * <p> + * A generic {@link org.apache.mahout.cf.taste.model.DataModel} designed for use with other JDBC data sources; + * one just specifies all necessary SQL queries to the constructor here. Optionally, the queries can be + * specified from a {@link Properties} object, {@link File}, or {@link InputStream}. This class is most + * appropriate when other existing implementations of {@link AbstractJDBCDataModel} are not suitable. If you + * are using this class to support a major database, consider contributing a specialized implementation of + * {@link AbstractJDBCDataModel} to the project for this database. + * </p> + */ +public final class GenericJDBCDataModel extends AbstractJDBCDataModel { + + public static final String DATA_SOURCE_KEY = "dataSource"; + public static final String GET_PREFERENCE_SQL_KEY = "getPreferenceSQL"; + public static final String GET_PREFERENCE_TIME_SQL_KEY = "getPreferenceTimeSQL"; + public static final String GET_USER_SQL_KEY = "getUserSQL"; + public static final String GET_ALL_USERS_SQL_KEY = "getAllUsersSQL"; + public static final String GET_NUM_USERS_SQL_KEY = "getNumUsersSQL"; + public static final String GET_NUM_ITEMS_SQL_KEY = "getNumItemsSQL"; + public static final String SET_PREFERENCE_SQL_KEY = "setPreferenceSQL"; + public static final String REMOVE_PREFERENCE_SQL_KEY = "removePreferenceSQL"; + public static final String GET_USERS_SQL_KEY = "getUsersSQL"; + public static final String GET_ITEMS_SQL_KEY = "getItemsSQL"; + public static final String GET_PREFS_FOR_ITEM_SQL_KEY = "getPrefsForItemSQL"; + public static final String GET_NUM_PREFERENCE_FOR_ITEM_KEY = "getNumPreferenceForItemSQL"; + public static final String GET_NUM_PREFERENCE_FOR_ITEMS_KEY = "getNumPreferenceForItemsSQL"; + public static final String GET_MAX_PREFERENCE_KEY = "getMaxPreferenceSQL"; + public static final String GET_MIN_PREFERENCE_KEY = "getMinPreferenceSQL"; + + /** + * <p> + * Specifies all SQL queries in a {@link Properties} object. See the {@code *_KEY} constants in this + * class (e.g. {@link #GET_USER_SQL_KEY}) for a list of all keys which must map to a value in this object. + * </p> + * + * @param props + * {@link Properties} object containing values + * @throws TasteException + * if anything goes wrong during initialization + */ + public GenericJDBCDataModel(Properties props) throws TasteException { + super(AbstractJDBCComponent.lookupDataSource(props.getProperty(DATA_SOURCE_KEY)), + props.getProperty(GET_PREFERENCE_SQL_KEY), + props.getProperty(GET_PREFERENCE_TIME_SQL_KEY), + props.getProperty(GET_USER_SQL_KEY), + props.getProperty(GET_ALL_USERS_SQL_KEY), + props.getProperty(GET_NUM_ITEMS_SQL_KEY), + props.getProperty(GET_NUM_USERS_SQL_KEY), + props.getProperty(SET_PREFERENCE_SQL_KEY), + props.getProperty(REMOVE_PREFERENCE_SQL_KEY), + props.getProperty(GET_USERS_SQL_KEY), + props.getProperty(GET_ITEMS_SQL_KEY), + props.getProperty(GET_PREFS_FOR_ITEM_SQL_KEY), + props.getProperty(GET_NUM_PREFERENCE_FOR_ITEM_KEY), + props.getProperty(GET_NUM_PREFERENCE_FOR_ITEMS_KEY), + props.getProperty(GET_MAX_PREFERENCE_KEY), + props.getProperty(GET_MIN_PREFERENCE_KEY)); + } + + /** + * <p> + * See {@link #GenericJDBCDataModel(Properties)}. This constructor reads values from a file + * instead, as if with {@link Properties#load(InputStream)}. So, the file should be in standard Java + * properties file format -- containing {@code key=value} pairs, one per line. + * </p> + * + * @param propertiesFile + * properties file + * @throws TasteException + * if anything goes wrong during initialization + */ + public GenericJDBCDataModel(File propertiesFile) throws TasteException { + this(getPropertiesFromFile(propertiesFile)); + } + + /** + * <p> + * See {@link #GenericJDBCDataModel(Properties)}. This constructor reads values from a resource available in + * the classpath, as if with {@link Class#getResourceAsStream(String)} and + * {@link Properties#load(InputStream)}. This is useful if your configuration file is, for example, packaged + * in a JAR file that is in the classpath. + * </p> + * + * @param resourcePath + * path to resource in classpath (e.g. "/com/foo/TasteSQLQueries.properties") + * @throws TasteException + * if anything goes wrong during initialization + */ + public GenericJDBCDataModel(String resourcePath) throws TasteException { + this(getPropertiesFromStream(GenericJDBCDataModel.class + .getResourceAsStream(resourcePath))); + } + + private static Properties getPropertiesFromFile(File file) throws TasteException { + try { + return getPropertiesFromStream(new FileInputStream(file)); + } catch (FileNotFoundException fnfe) { + throw new TasteException(fnfe); + } + } + + private static Properties getPropertiesFromStream(InputStream is) throws TasteException { + try { + try { + Properties props = new Properties(); + props.load(is); + return props; + } finally { + Closeables.close(is, true); + } + } catch (IOException ioe) { + throw new TasteException(ioe); + } + } + +}
